scottk/siembol
1
0
Fork 0
mirror of https://github.com/optim-enterprises-bv/siembol.git synced 2025-12-01 17:53:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

moving num workers into parsing application (#689)

Browse Source 
fixing header reoutingapp with null headers
This commit is contained in:
Marian Novotny
2022-06-21 12:33:56 +01:00
committed by GitHub
parent 71830b8c43
commit 6111335983
30 changed files with 113 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
parsing/parsing-app/src/main/java/uk/co/gresearch/siembol/parsers/application/factory/ParsingApplicationFactoryAttributes.java
View File

@@ -9,6 +9,7 @@ public class ParsingApplicationFactoryAttributes {
private String jsonSchema;
private String name;
private String applicationParserSpecification;
private Integer numWorkers;
private Integer inputParallelism;
private Integer outputParallelism;
private Integer parsingParallelism;
@@ -105,4 +106,12 @@ public class ParsingApplicationFactoryAttributes {
public void setApplicationType(ParsingApplicationTypeDto applicationType) {
this.applicationType = applicationType;
}
public Integer getNumWorkers() {
return numWorkers;
}
public void setNumWorkers(Integer numWorkers) {
this.numWorkers = numWorkers;
}
}

1
parsing/parsing-app/src/main/java/uk/co/gresearch/siembol/parsers/application/factory/ParsingApplicationFactoryImpl.java
View File

@@ -330,6 +330,7 @@ public class ParsingApplicationFactoryImpl implements ParsingApplicationFactory
attributes.setSourceHeaderName(
application.getParsingSettingsDto().getHeaderRoutingParserDto().getHeaderName());
}
attributes.setNumWorkers(application.getParsingApplicationSettingsDto().getNumWorkers());
}
private String getParserFromMap(String parserName, Map<String, String> parserMap) {

17
parsing/parsing-app/src/main/java/uk/co/gresearch/siembol/parsers/application/model/ParsingApplicationSettingsDto.java
View File

@@ -19,18 +19,21 @@ public class ParsingApplicationSettingsDto {
@Attributes(description = "The kafka topic for publishing error messages", required = true)
private String errorTopic;
@JsonProperty("num_workers")
@Attributes(description = "The number of workers for the parsing application", minimum = 1, required = true)
private Integer numWorkers = 1;
@JsonProperty("input_parallelism")
@Attributes(description = "The number of parallel executors for reading messages from the input kafka topics",
@Attributes(description = "The number of parallel executors per worker for reading messages from the input topics",
required = true, minimum = 1)
private Integer inputParallelism;
@JsonProperty("parsing_parallelism")
@Attributes(description = "The number of parallel executors for parsing messages",
@Attributes(description = "The number of parallel executors per worker for parsing messages",
required = true, minimum = 1)
private Integer parsingParallelism;
@JsonProperty("output_parallelism")
@Attributes(description = "The number of parallel executors for publishing to kafka",
@Attributes(description = "The number of parallel executors per worker for publishing to kafka",
required = true, minimum = 1)
private Integer outputParallelism;
@@ -141,4 +144,12 @@ public class ParsingApplicationSettingsDto {
public void setOriginalStringTopic(String originalStringTopic) {
this.originalStringTopic = originalStringTopic;
}
public Integer getNumWorkers() {
return numWorkers;
}
public void setNumWorkers(Integer numWorkers) {
this.numWorkers = numWorkers;
}
}

8
parsing/parsing-app/src/test/java/uk/co/gresearch/siembol/parsers/application/factory/ParsingApplicationFactoryImplTest.java
View File

@@ -24,6 +24,7 @@ public class ParsingApplicationFactoryImplTest {
"max_num_fields" : 100,
"max_field_size" : 40000,
"original_string_topic" : "truncated",
"num_workers" : 3,
"input_parallelism": 1,
"parsing_parallelism": 2,
"output_parallelism": 3,
@@ -49,6 +50,7 @@ public class ParsingApplicationFactoryImplTest {
"secret"
],
"error_topic": "error",
"num_workers" : 3,
"input_parallelism": 1,
"parsing_parallelism": 2,
"output_parallelism": 3,
@@ -96,6 +98,7 @@ public class ParsingApplicationFactoryImplTest {
"public"
],
"error_topic": "error",
"num_workers" : 3,
"input_parallelism": 1,
"parsing_parallelism": 2,
"output_parallelism": 3,
@@ -144,6 +147,7 @@ public class ParsingApplicationFactoryImplTest {
"public"
],
"error_topic": "error",
"num_workers" : 3,
"input_parallelism": 1,
"parsing_parallelism": 2,
"output_parallelism": 3,
@@ -260,6 +264,7 @@ public class ParsingApplicationFactoryImplTest {
ParsingApplicationFactoryResult result = factory.create(simpleSingleApplicationParser, testParsersConfigs);
Assert.assertSame(ParsingApplicationFactoryResult.StatusCode.OK, result.getStatusCode());
Assert.assertEquals("test", result.getAttributes().getName());
Assert.assertEquals(3, result.getAttributes().getNumWorkers().intValue());
Assert.assertEquals(1, result.getAttributes().getInputParallelism().intValue());
Assert.assertEquals(2, result.getAttributes().getParsingParallelism().intValue());
Assert.assertEquals(3, result.getAttributes().getOutputParallelism().intValue());
@@ -312,6 +317,7 @@ public class ParsingApplicationFactoryImplTest {
ParsingApplicationFactoryResult result = factory.create(simpleRoutingApplicationParser, testParsersConfigs);
Assert.assertSame(ParsingApplicationFactoryResult.StatusCode.OK, result.getStatusCode());
Assert.assertEquals("test", result.getAttributes().getName());
Assert.assertEquals(3, result.getAttributes().getNumWorkers().intValue());
Assert.assertEquals(1, result.getAttributes().getInputParallelism().intValue());
Assert.assertEquals(2, result.getAttributes().getParsingParallelism().intValue());
Assert.assertEquals(3, result.getAttributes().getOutputParallelism().intValue());
@@ -397,6 +403,7 @@ public class ParsingApplicationFactoryImplTest {
ParsingApplicationFactoryResult result = factory.create(headerRoutingApplicationParser, testParsersConfigs);
Assert.assertSame(ParsingApplicationFactoryResult.StatusCode.OK, result.getStatusCode());
Assert.assertEquals("test", result.getAttributes().getName());
Assert.assertEquals(3, result.getAttributes().getNumWorkers().intValue());
Assert.assertEquals(1, result.getAttributes().getInputParallelism().intValue());
Assert.assertEquals(2, result.getAttributes().getParsingParallelism().intValue());
Assert.assertEquals(3, result.getAttributes().getOutputParallelism().intValue());
@@ -452,6 +459,7 @@ public class ParsingApplicationFactoryImplTest {
ParsingApplicationFactoryResult result = factory.create(topicRoutingApplicationParser, testParsersConfigs);
Assert.assertSame(ParsingApplicationFactoryResult.StatusCode.OK, result.getStatusCode());
Assert.assertEquals("test", result.getAttributes().getName());
Assert.assertEquals(3, result.getAttributes().getNumWorkers().intValue());
Assert.assertEquals(1, result.getAttributes().getInputParallelism().intValue());
Assert.assertEquals(2, result.getAttributes().getParsingParallelism().intValue());
Assert.assertEquals(3, result.getAttributes().getOutputParallelism().intValue());