This is an automated email from the ASF dual-hosted git repository.
chrisdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x-extras.git
The following commit(s) were added to refs/heads/develop by this push:
new d49cdbd6 Chore/update asf yml (#757)
d49cdbd6 is described below
commit d49cdbd65b4db5594ff9fcd3958197890d25f13d
Author: Christofer Dutz <[email protected]>
AuthorDate: Thu Aug 13 18:03:46 2026 +0200
Chore/update asf yml (#757)
* refactor: Updated the kafka and calcite modules to use the new event-pump
instead of the now obsolete scraper.
* chore: Updated the .asf.yml to match that of the main repo.
---
.asf.yaml | 5 +-
plc4j/integrations/apache-calcite/pom.xml | 2 +-
.../main/java/org/apache/plc4x/Plc4xBaseTable.java | 8 +-
.../main/java/org/apache/plc4x/Plc4xSchema.java | 64 +++---
.../java/org/apache/plc4x/Plc4xSchemaFactory.java | 21 +-
.../java/org/apache/plc4x/Plc4xStreamTable.java | 4 +-
.../src/main/java/org/apache/plc4x/Plc4xTable.java | 4 +-
.../java/org/apache/plc4x/DriverManagerTest.java | 9 +-
.../java/org/apache/plc4x/Plc4XBaseTableTest.java | 12 +-
.../apache-calcite/src/test/resources/example.yml | 30 ++-
plc4j/integrations/apache-kafka/README.md | 3 +-
.../apache-kafka/config/plc4x-source.properties | 4 +-
plc4j/integrations/apache-kafka/pom.xml | 2 +-
.../apache/plc4x/kafka/Plc4xSourceConnector.java | 6 +-
.../org/apache/plc4x/kafka/Plc4xSourceTask.java | 216 ++++++++++++---------
.../apache/plc4x/kafka/config/SinkTaskTest.java | 2 +-
.../apache/plc4x/kafka/config/SourceTaskTest.java | 17 +-
.../kafka/properties/sink_task_no_error.properties | 22 +--
.../properties/source_task_no_error.properties | 22 +--
19 files changed, 257 insertions(+), 196 deletions(-)
diff --git a/.asf.yaml b/.asf.yaml
index 4a1d4e77..322a356e 100644
--- a/.asf.yaml
+++ b/.asf.yaml
@@ -76,14 +76,15 @@ github:
excludes: []
bypass_teams:
- root
+ - plc4x-committers
restrict_deletion: true
restrict_force_push: true
required_pull_request_reviews:
dismiss_stale_reviews: true
- require_last_push_approval: false
+ require_last_push_approval: true
required_approving_review_count: 1
required_linear_history: false
- required_signatures: true
+ required_signatures: false
required_conversation_resolution: true
notifications:
diff --git a/plc4j/integrations/apache-calcite/pom.xml
b/plc4j/integrations/apache-calcite/pom.xml
index 9b2b79ac..c794faee 100644
--- a/plc4j/integrations/apache-calcite/pom.xml
+++ b/plc4j/integrations/apache-calcite/pom.xml
@@ -57,7 +57,7 @@
</dependency>
<dependency>
<groupId>org.apache.plc4x</groupId>
- <artifactId>plc4j-scraper</artifactId>
+ <artifactId>plc4j-tools-event-pump</artifactId>
<version>${plc4x.version}</version>
</dependency>
<dependency>
diff --git
a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xBaseTable.java
b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xBaseTable.java
index 625e4d39..445c185e 100644
---
a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xBaseTable.java
+++
b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xBaseTable.java
@@ -29,7 +29,7 @@ import org.apache.calcite.schema.Statistic;
import org.apache.calcite.schema.impl.AbstractTable;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.plc4x.java.api.exceptions.PlcRuntimeException;
-import org.apache.plc4x.java.scraper.config.JobConfiguration;
+import org.apache.plc4x.java.tools.eventpump.config.BatchConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -53,12 +53,12 @@ public abstract class Plc4xBaseTable extends AbstractTable {
private static final Logger logger =
LoggerFactory.getLogger(Plc4xBaseTable.class);
private final BlockingQueue<Plc4xSchema.Record> queue;
- private final JobConfiguration conf;
+ private final BatchConfiguration conf;
private final long tableCutoff;
private Plc4xSchema.Record current;
private final List<String> names;
- public Plc4xBaseTable(BlockingQueue<Plc4xSchema.Record> queue,
JobConfiguration conf, long tableCutoff) {
+ public Plc4xBaseTable(BlockingQueue<Plc4xSchema.Record> queue,
BatchConfiguration conf, long tableCutoff) {
this.tableCutoff = tableCutoff;
logger.info("Instantiating new PLC4X Table with configuration: {}",
conf);
this.queue = queue;
@@ -107,7 +107,7 @@ public abstract class Plc4xBaseTable extends AbstractTable {
} catch (ExecutionException | TimeoutException e) {
throw new PlcRuntimeException("Unable to fetch first record and
infer arguments!", e);
}
- logger.info("Inferring types for Table '{}' based on values: {}",
conf.getName(), first.values);
+ logger.info("Inferring types for Table '{}' based on values: {}",
conf.getId(), first.values);
// Extract types
List<RelDataType> types = names.stream()
.map(n -> {
diff --git
a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchema.java
b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchema.java
index 9f1b3ee3..6950a505 100644
---
a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchema.java
+++
b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchema.java
@@ -22,12 +22,12 @@ import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.impl.AbstractSchema;
import org.apache.plc4x.java.DefaultPlcDriverManager;
import org.apache.plc4x.java.api.exceptions.PlcRuntimeException;
-import org.apache.plc4x.java.scraper.ResultHandler;
-import org.apache.plc4x.java.scraper.Scraper;
-import org.apache.plc4x.java.scraper.ScraperImpl;
-import org.apache.plc4x.java.scraper.config.JobConfiguration;
-import org.apache.plc4x.java.scraper.config.ScraperConfiguration;
-import org.apache.plc4x.java.scraper.exception.ScraperException;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.tools.eventpump.EventPump;
+import org.apache.plc4x.java.tools.eventpump.TagBatch;
+import org.apache.plc4x.java.tools.eventpump.config.BatchConfiguration;
+import org.apache.plc4x.java.tools.eventpump.config.EventPumpConfiguration;
+import org.apache.plc4x.java.tools.eventpump.config.EventPumpFactory;
import org.apache.plc4x.java.utils.cache.CachedPlcConnectionManager;
import java.time.Instant;
@@ -38,36 +38,43 @@ import java.util.stream.Collectors;
public class Plc4xSchema extends AbstractSchema {
- protected final ScraperConfiguration configuration;
- protected final Scraper scraper;
+ protected final EventPumpConfiguration configuration;
+ protected final EventPump eventPump;
protected final QueueHandler handler;
protected final Map<String, BlockingQueue<Record>> queues;
protected final Map<String, Table> tableMap;
+ /** batch id -> connection id, so a record can be attributed to the PLC
it came from. */
+ protected final Map<String, String> connectionIds;
- public Plc4xSchema(ScraperConfiguration configuration, long tableCutoff)
throws ScraperException {
+ public Plc4xSchema(EventPumpConfiguration configuration, long tableCutoff)
throws Exception {
this.configuration = configuration;
this.handler = new QueueHandler();
- this.scraper = new ScraperImpl(handler,
- CachedPlcConnectionManager.getBuilder()
- .withConnectionManager(new DefaultPlcDriverManager())
- .build(),
- configuration.getJobs());
- this.queues = configuration.getJobConfigurations().stream()
+ this.connectionIds = configuration.getBatches().stream()
+ .collect(Collectors.toMap(
+ BatchConfiguration::getId,
+ BatchConfiguration::getConnectionId
+ ));
+ this.queues = configuration.getBatches().stream()
.collect(Collectors.toMap(
- JobConfiguration::getName,
+ BatchConfiguration::getId,
conf -> new ArrayBlockingQueue<>(1000)
));
- // Create the tables
- this.tableMap = configuration.getJobConfigurations().stream()
+ // Create the tables - one per batch
+ this.tableMap = configuration.getBatches().stream()
.collect(Collectors.toMap(
- JobConfiguration::getName,
- conf -> defineTable(queues.get(conf.getName()), conf,
tableCutoff)
+ BatchConfiguration::getId,
+ conf -> defineTable(queues.get(conf.getId()), conf,
tableCutoff)
));
- // Start the scraper
- this.scraper.start();
+ // Every batch reports to the same handler, which routes by batch id
+ this.eventPump = EventPumpFactory.create(configuration,
+ CachedPlcConnectionManager.getBuilder()
+ .withConnectionManager(new DefaultPlcDriverManager())
+ .build(),
+ handler);
+ this.eventPump.startAll();
}
- Table defineTable(BlockingQueue<Record> queue, JobConfiguration
configuration, Long limit) {
+ Table defineTable(BlockingQueue<Record> queue, BatchConfiguration
configuration, Long limit) {
if (limit <= 0) {
return new Plc4xStreamTable(queue, configuration);
} else {
@@ -94,13 +101,16 @@ public class Plc4xSchema extends AbstractSchema {
}
}
- class QueueHandler implements ResultHandler {
+ class QueueHandler implements TagBatch.TagBatchListener {
@Override
- public void handle(String job, String alias, Map<String, Object>
results) {
+ public void onTagsFetched(TagBatch batch, PlcReadResponse response) {
+ String batchId = batch.getBatchId();
+ Map<String, Object> results = response.getTagNames().stream()
+ .collect(Collectors.toMap(name -> name, response::getObject));
try {
- Record record = new Record(Instant.now(), alias, results);
- queues.get(job).put(record);
+ Record record = new Record(Instant.now(),
connectionIds.get(batchId), results);
+ queues.get(batchId).put(record);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new PlcRuntimeException("Handling got interrupted", e);
diff --git
a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchemaFactory.java
b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchemaFactory.java
index b2876107..92e4d2e5 100644
---
a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchemaFactory.java
+++
b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xSchemaFactory.java
@@ -22,12 +22,11 @@ import org.apache.calcite.schema.Schema;
import org.apache.calcite.schema.SchemaFactory;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.commons.lang3.Validate;
-import org.apache.plc4x.java.scraper.config.ScraperConfiguration;
-import
org.apache.plc4x.java.scraper.config.triggeredscraper.ScraperConfigurationTriggeredImpl;
-import org.apache.plc4x.java.scraper.exception.ScraperException;
+import org.apache.plc4x.java.tools.eventpump.config.EventPumpConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
import java.io.IOException;
import java.util.Map;
@@ -40,9 +39,17 @@ public class Plc4xSchemaFactory implements SchemaFactory {
Object config = operand.get("config");
Validate.notNull(config, "No configuration file given. Please specify
operand 'config'...'");
// Load configuration from file
- ScraperConfiguration configuration;
+ EventPumpConfiguration configuration;
+ String configPath = config.toString();
try {
- configuration = ScraperConfiguration.fromFile(config.toString(),
ScraperConfigurationTriggeredImpl.class);
+ File configFile = new File(configPath);
+ if (configPath.endsWith(".json")) {
+ configuration = EventPumpConfiguration.fromJson(configFile);
+ } else if (configPath.endsWith(".xml")) {
+ configuration = EventPumpConfiguration.fromXml(configFile);
+ } else {
+ configuration = EventPumpConfiguration.fromYaml(configFile);
+ }
} catch (IOException e) {
throw new IllegalArgumentException("Unable to load configuration
file!", e);
}
@@ -59,8 +66,8 @@ public class Plc4xSchemaFactory implements SchemaFactory {
// Pass the configuration to the Schema
try {
return new Plc4xSchema(configuration, parsedLimit);
- } catch (ScraperException e) {
- LOGGER.warn("Could not evaluate Plc4xSchema",e);
+ } catch (Exception e) {
+ LOGGER.warn("Could not evaluate Plc4xSchema", e);
//ToDo Exception, but interface does not accept ... null is fishy
return null;
}
diff --git
a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xStreamTable.java
b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xStreamTable.java
index 7c9668d9..840c8cac 100644
---
a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xStreamTable.java
+++
b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xStreamTable.java
@@ -23,13 +23,13 @@ import org.apache.calcite.linq4j.Enumerable;
import org.apache.calcite.schema.ScannableTable;
import org.apache.calcite.schema.StreamableTable;
import org.apache.calcite.schema.Table;
-import org.apache.plc4x.java.scraper.config.JobConfiguration;
+import org.apache.plc4x.java.tools.eventpump.config.BatchConfiguration;
import java.util.concurrent.BlockingQueue;
public class Plc4xStreamTable extends Plc4xBaseTable implements
ScannableTable, StreamableTable {
- public Plc4xStreamTable(BlockingQueue<Plc4xSchema.Record> queue,
JobConfiguration conf) {
+ public Plc4xStreamTable(BlockingQueue<Plc4xSchema.Record> queue,
BatchConfiguration conf) {
super(queue, conf, -1L);
}
diff --git
a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xTable.java
b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xTable.java
index 9c479a03..2b64b682 100644
---
a/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xTable.java
+++
b/plc4j/integrations/apache-calcite/src/main/java/org/apache/plc4x/Plc4xTable.java
@@ -21,13 +21,13 @@ package org.apache.plc4x;
import org.apache.calcite.DataContext;
import org.apache.calcite.linq4j.Enumerable;
import org.apache.calcite.schema.ScannableTable;
-import org.apache.plc4x.java.scraper.config.JobConfiguration;
+import org.apache.plc4x.java.tools.eventpump.config.BatchConfiguration;
import java.util.concurrent.BlockingQueue;
public class Plc4xTable extends Plc4xBaseTable implements ScannableTable {
- public Plc4xTable(BlockingQueue<Plc4xSchema.Record> queue,
JobConfiguration conf, long tableCutoff) {
+ public Plc4xTable(BlockingQueue<Plc4xSchema.Record> queue,
BatchConfiguration conf, long tableCutoff) {
super(queue, conf, tableCutoff);
}
diff --git
a/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/DriverManagerTest.java
b/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/DriverManagerTest.java
index c9f93233..8116acd0 100644
---
a/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/DriverManagerTest.java
+++
b/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/DriverManagerTest.java
@@ -20,12 +20,11 @@ package org.apache.plc4x;
import org.apache.calcite.jdbc.CalciteConnection;
import org.apache.calcite.jdbc.Driver;
-import org.apache.plc4x.java.scraper.config.ScraperConfiguration;
-import org.apache.plc4x.java.scraper.config.ScraperConfigurationClassicImpl;
-import org.apache.plc4x.java.scraper.exception.ScraperException;
+import org.apache.plc4x.java.tools.eventpump.config.EventPumpConfiguration;
import org.assertj.core.api.WithAssertions;
import org.junit.jupiter.api.Test;
+import java.io.File;
import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
@@ -36,12 +35,12 @@ import java.util.Properties;
public class DriverManagerTest implements WithAssertions {
@Test
- void query() throws SQLException, IOException, ScraperException {
+ void query() throws Exception {
Driver driver = new Driver();
Connection connection =
driver.connect("jdbc:calcite:asdf;lex=MYSQL_ANSI", new Properties());
CalciteConnection calciteConnection =
connection.unwrap(CalciteConnection.class);
- calciteConnection.getRootSchema().add("plc4x", new
Plc4xSchema(ScraperConfiguration.fromFile("src/test/resources/example.yml",
ScraperConfigurationClassicImpl.class), 100));
+ calciteConnection.getRootSchema().add("plc4x", new
Plc4xSchema(EventPumpConfiguration.fromYaml(new
File("src/test/resources/example.yml")), 100));
ResultSet rs = connection.prepareStatement("SELECT * FROM
\"plc4x\".\"job1\"").executeQuery();
validateResult(rs);
diff --git
a/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/Plc4XBaseTableTest.java
b/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/Plc4XBaseTableTest.java
index 19c719a1..932e7ae0 100644
---
a/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/Plc4XBaseTableTest.java
+++
b/plc4j/integrations/apache-calcite/src/test/java/org/apache/plc4x/Plc4XBaseTableTest.java
@@ -19,7 +19,7 @@
package org.apache.plc4x;
import org.apache.calcite.linq4j.Enumerator;
-import org.apache.plc4x.java.scraper.config.JobConfigurationImpl;
+import org.apache.plc4x.java.tools.eventpump.config.BatchConfiguration;
import org.assertj.core.api.WithAssertions;
import org.junit.jupiter.api.Test;
@@ -33,12 +33,10 @@ class Plc4XBaseTableTest implements WithAssertions {
@Test
void testOnBlockingQueue() {
ArrayBlockingQueue<Plc4xSchema.Record> queue = new
ArrayBlockingQueue<>(100);
- Plc4xStreamTable table = new Plc4xStreamTable(queue, new
JobConfigurationImpl(
- "job1",
- null,
- 100,
- Collections.emptyList(),
- Collections.singletonMap("key", "address")));
+ BatchConfiguration conf = new BatchConfiguration();
+ conf.setId("job1");
+ conf.setSimpleTags(Collections.singletonMap("key", "address"));
+ Plc4xStreamTable table = new Plc4xStreamTable(queue, conf);
Map<String, Object> objects = Collections.singletonMap("key", "value");
queue.add(new Plc4xSchema.Record(Instant.now(), "", objects));
diff --git a/plc4j/integrations/apache-calcite/src/test/resources/example.yml
b/plc4j/integrations/apache-calcite/src/test/resources/example.yml
index d90dd3c1..b0809a17 100644
--- a/plc4j/integrations/apache-calcite/src/test/resources/example.yml
+++ b/plc4j/integrations/apache-calcite/src/test/resources/example.yml
@@ -17,16 +17,28 @@
# under the License.
# ----------------------------------------------------------------------------
---
-sources:
- test: simulated:test
- test2: simulated:test2
+connections:
+ - id: test
+ url: simulated:test
+ - id: test2
+ url: simulated:test2
-jobs:
- - name: job1
- scrapeRate: 10
- sources:
- - test
- - test2
+# Every batch becomes one table, named after the batch id.
+batches:
+ - id: job1
+ connectionId: test
tags:
test: 'RANDOM/test:DINT'
test2: 'RANDOM/test:STRING'
+ trigger:
+ type: timer
+ intervalMillis: 10
+
+ - id: job2
+ connectionId: test2
+ tags:
+ test: 'RANDOM/test:DINT'
+ test2: 'RANDOM/test:STRING'
+ trigger:
+ type: timer
+ intervalMillis: 10
diff --git a/plc4j/integrations/apache-kafka/README.md
b/plc4j/integrations/apache-kafka/README.md
index b8c5be77..13dbdbf5 100644
--- a/plc4j/integrations/apache-kafka/README.md
+++ b/plc4j/integrations/apache-kafka/README.md
@@ -203,7 +203,8 @@ If an error occurs when reading or writing PLC addresses a
graceful backoff has
bombarded with requests. However as the number of connectors for each PLC
should be limited to reduce the load on the PLC,
the graceful backoff shouldn't have a major impact.
-For the source connector the PLC4X scraper logic is able to handle randomized
polling rates on failures, this is buffered within the
+For the source connector the PLC4X event-pump backs off exponentially on
failures (1s, doubling up to 60s) and skips a poll
+if the previous read is still running, this is buffered within the
connector, the poll rate of the connector has no affect on the PLC poll rate.
For the sink connector, if a write fails it is retried a configurable number
of times with a timeout between each time.
diff --git a/plc4j/integrations/apache-kafka/config/plc4x-source.properties
b/plc4j/integrations/apache-kafka/config/plc4x-source.properties
index 6c024cb7..13205b99 100644
--- a/plc4j/integrations/apache-kafka/config/plc4x-source.properties
+++ b/plc4j/integrations/apache-kafka/config/plc4x-source.properties
@@ -32,7 +32,7 @@ sources.machineA.connectionString=simulated://127.0.0.1
#This value controls how often it returns when no messages are received.
sources.machineA.pollReturnInterval=5000
-#There is an internal buffer between the PLC4X scraper and Kafka Connect.
+#There is an internal buffer between the PLC4X event-pump and Kafka Connect.
#This is the size of that buffer.
sources.machineA.bufferSize=1000
@@ -45,7 +45,7 @@
sources.machineA.jobReferences.simulated-heartbeat.topic=simulated-heartbeat-top
#A list of jobs specified in the following section.
jobs=simulated-dashboard,simulated-heartbeat
-#The poll rate for this job. the PLC4X scraper will request data every
interval (ms).
+#The poll rate for this job. The PLC4X event-pump will request data every
interval (ms).
jobs.simulated-dashboard.interval=1000
#A list of tags. Each tag is a map between an alias and a PLC4X address.
diff --git a/plc4j/integrations/apache-kafka/pom.xml
b/plc4j/integrations/apache-kafka/pom.xml
index 18bc4bdd..8d9ecc00 100644
--- a/plc4j/integrations/apache-kafka/pom.xml
+++ b/plc4j/integrations/apache-kafka/pom.xml
@@ -157,7 +157,7 @@
</dependency>
<dependency>
<groupId>org.apache.plc4x</groupId>
- <artifactId>plc4j-scraper</artifactId>
+ <artifactId>plc4j-tools-event-pump</artifactId>
<version>${plc4x.version}</version>
</dependency>
<dependency>
diff --git
a/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceConnector.java
b/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceConnector.java
index 55270756..86731a24 100644
---
a/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceConnector.java
+++
b/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceConnector.java
@@ -57,7 +57,7 @@ public class Plc4xSourceConnector extends SourceConnector {
@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
// Initially we planned to have the simple assumption that one task
maps to one PLC connection.
- // But we could easily say that one scraper instance maps to a task
and one scraper task can
+ // But we could easily say that one event-pump instance maps to a task
and one batch can
// process multiple PLC connections. But I guess this would be an
optimization as we have to
// balance the load manually.
if(sourceConfig.getJobs().size() > maxTasks) {
@@ -66,8 +66,8 @@ public class Plc4xSourceConnector extends SourceConnector {
return Collections.emptyList();
}
- // For each configured source we'll start a dedicated scraper instance
collecting
- // all the scraper jobs enabled for this source.
+ // For each configured source we'll start a dedicated event-pump
instance collecting
+ // all the jobs enabled for this source.
List<Map<String, String>> configs = new LinkedList<>();
for (Source source : sourceConfig.getSources()) {
// Build a list of job configurations only containing the ones
referenced from
diff --git
a/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceTask.java
b/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceTask.java
index 9a4a3538..3bc3f7a8 100644
---
a/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceTask.java
+++
b/plc4j/integrations/apache-kafka/src/main/java/org/apache/plc4x/kafka/Plc4xSourceTask.java
@@ -28,13 +28,10 @@ import org.apache.kafka.connect.source.SourceTask;
import org.apache.plc4x.java.DefaultPlcDriverManager;
import org.apache.plc4x.java.api.PlcConnectionManager;
import org.apache.plc4x.java.api.value.PlcValue;
-import
org.apache.plc4x.java.scraper.config.triggeredscraper.JobConfigurationTriggeredImplBuilder;
-import
org.apache.plc4x.java.scraper.config.triggeredscraper.ScraperConfigurationTriggeredImpl;
-import
org.apache.plc4x.java.scraper.config.triggeredscraper.ScraperConfigurationTriggeredImplBuilder;
-import org.apache.plc4x.java.scraper.exception.ScraperException;
-import org.apache.plc4x.java.scraper.triggeredscraper.TriggeredScraperImpl;
-import
org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.collector.TriggerCollector;
-import
org.apache.plc4x.java.scraper.triggeredscraper.triggerhandler.collector.TriggerCollectorImpl;
+import org.apache.plc4x.java.api.messages.PlcReadResponse;
+import org.apache.plc4x.java.tools.eventpump.EventPump;
+import org.apache.plc4x.java.tools.eventpump.TagBatch;
+import org.apache.plc4x.java.tools.eventpump.triggers.TimerTrigger;
import org.apache.plc4x.java.utils.cache.CachedPlcConnectionManager;
import org.apache.plc4x.kafka.config.Constants;
import org.apache.plc4x.kafka.util.VersionUtil;
@@ -93,10 +90,10 @@ public class Plc4xSourceTask extends SourceTask {
.field(Constants.JOB_NAME_FIELD, Schema.STRING_SCHEMA)
.build();
- // Internal buffer into which all incoming scraper responses are written
to.
+ // Internal buffer into which all incoming responses are written to.
private ArrayBlockingQueue<SourceRecord> buffer;
private Integer pollReturnInterval;
- private TriggeredScraperImpl scraper;
+ private EventPump eventPump;
private final SecureRandom random = new SecureRandom();
@Override
@@ -112,12 +109,14 @@ public class Plc4xSourceTask extends SourceTask {
pollReturnInterval = config.getInt(Constants.KAFKA_POLL_RETURN_CONFIG);
Integer bufferSize = config.getInt(Constants.BUFFER_SIZE_CONFIG);
- Map<String, String> topics = new HashMap<>();
// Create a buffer with a capacity of BUFFER_SIZE_CONFIG elements
which schedules access in a fair way.
buffer = new ArrayBlockingQueue<>(bufferSize, true);
- ScraperConfigurationTriggeredImplBuilder builder = new
ScraperConfigurationTriggeredImplBuilder();
- builder.addSource(connectionName, plc4xConnectionString);
+ PlcConnectionManager connectionManager =
CachedPlcConnectionManager.getBuilder()
+ .withConnectionManager(new DefaultPlcDriverManager())
+ .build();
+
+ eventPump = new EventPump();
List<String> jobConfigs = config.getList(Constants.QUERIES_CONFIG);
for (String jobConfig : jobConfigs) {
@@ -129,11 +128,11 @@ public class Plc4xSourceTask extends SourceTask {
continue;
}
- String jobName = jobConfigSegments[0];
- String topic = jobConfigSegments[1];
- Integer rate = Integer.valueOf(jobConfigSegments[2]);
- JobConfigurationTriggeredImplBuilder jobBuilder = builder.job(
- jobName, String.format("(SCHEDULED,%s)",
rate)).source(connectionName);
+ final String jobName = jobConfigSegments[0];
+ final String topic = jobConfigSegments[1];
+ final int rate = Integer.parseInt(jobConfigSegments[2]);
+
+ Map<String, String> tags = new LinkedHashMap<>();
for (int i = 3; i < jobConfigSegments.length; i++) {
String[] tagSegments = jobConfigSegments[i].split("#");
if (tagSegments.length != 2) {
@@ -142,108 +141,131 @@ public class Plc4xSourceTask extends SourceTask {
jobName, jobConfigSegments[i]);
continue;
}
- String tagAlias = tagSegments[0];
- String tagAddress = tagSegments[1];
- jobBuilder.tag(tagAlias, tagAddress);
- topics.put(jobName, topic);
+ tags.put(tagSegments[0], tagSegments[1]);
+ }
+ if (tags.isEmpty()) {
+ log.warn("Job configuration '{}' doesn't contain any valid
tags ... skipping it.", jobName);
+ continue;
}
- jobBuilder.build();
- }
- ScraperConfigurationTriggeredImpl scraperConfig = builder.build();
+ // One batch per job: all tags of a job are read together, at the
job's rate.
+ TagBatch batch = TagBatch.builder()
+ .withBatchId(jobName)
+ .withConnectionManager(connectionManager)
+ .withConnectionString(plc4xConnectionString)
+ .addTagAddresses(tags)
+ .withTrigger(new TimerTrigger(rate, TimeUnit.MILLISECONDS))
+ .withListener(new TagBatch.TagBatchListener() {
+ @Override
+ public void onTagsFetched(TagBatch tagBatch,
PlcReadResponse response) {
+ handleResponse(tagBatch.getBatchId(), connectionName,
topic, response);
+ }
- try {
- PlcConnectionManager connectionManager =
CachedPlcConnectionManager.getBuilder()
- .withConnectionManager(new DefaultPlcDriverManager())
+ @Override
+ public void onError(TagBatch tagBatch, Throwable error) {
+ log.error("Error reading tags for job '{}': {}",
tagBatch.getBatchId(), error.getMessage());
+ }
+
+ @Override
+ public void onFetchSkipped(TagBatch tagBatch, long
lastFetchDurationMs, long consecutiveSkips) {
+ log.warn("Job '{}' is configured to be read every
{}ms, but the last read took {}ms " +
+ "({} consecutive reads skipped).",
+ tagBatch.getBatchId(), rate, lastFetchDurationMs,
consecutiveSkips);
+ }
+ })
.build();
- TriggerCollector triggerCollector = new
TriggerCollectorImpl(connectionManager);
- scraper = new TriggeredScraperImpl(scraperConfig,
connectionManager, (jobName, sourceName, results) -> {
- try {
- Long timestamp = System.currentTimeMillis();
+ eventPump.addBatch(batch);
+ }
- Map<String, String> sourcePartition = new HashMap<>();
- sourcePartition.put("sourceName", sourceName);
- sourcePartition.put("jobName", jobName);
+ eventPump.startAll();
+ }
- Map<String, Long> sourceOffset =
Collections.singletonMap("offset", timestamp);
+ /**
+ * Turns one response into a Kafka {@link SourceRecord} and adds it to the
buffer that
+ * {@link #poll()} drains.
+ */
+ private void handleResponse(String jobName, String sourceName, String
topic, PlcReadResponse response) {
+ Map<String, Object> results = response.getTagNames().stream()
+ .collect(HashMap::new, (map, name) -> map.put(name,
response.getObject(name)), HashMap::putAll);
+ try {
+ Long timestamp = System.currentTimeMillis();
- String topic = topics.get(jobName);
+ Map<String, String> sourcePartition = new HashMap<>();
+ sourcePartition.put("sourceName", sourceName);
+ sourcePartition.put("jobName", jobName);
- // Prepare the key structure.
- Struct key = new Struct(KEY_SCHEMA)
- .put(Constants.SOURCE_NAME_FIELD, sourceName)
- .put(Constants.JOB_NAME_FIELD, jobName);
+ Map<String, Long> sourceOffset =
Collections.singletonMap("offset", timestamp);
- // Build the Schema for the result struct.
- SchemaBuilder tagSchemaBuilder = SchemaBuilder.struct()
- .name("org.apache.plc4x.kafka.schema.Tag");
+ // Prepare the key structure.
+ Struct key = new Struct(KEY_SCHEMA)
+ .put(Constants.SOURCE_NAME_FIELD, sourceName)
+ .put(Constants.JOB_NAME_FIELD, jobName);
+ // Build the Schema for the result struct.
+ SchemaBuilder tagSchemaBuilder = SchemaBuilder.struct()
+ .name("org.apache.plc4x.kafka.schema.Tag");
- for (Map.Entry<String, Object> result :
results.entrySet()) {
- // Get tag-name and -value from the results.
- String tagName = result.getKey();
- Object tagValue = result.getValue();
- // Get the schema for the given value type.
- Schema valueSchema = getSchema(tagValue);
+ for (Map.Entry<String, Object> result : results.entrySet()) {
+ // Get tag-name and -value from the results.
+ String tagName = result.getKey();
+ Object tagValue = result.getValue();
- // Add the schema description for the current tag.
- tagSchemaBuilder.field(tagName, valueSchema);
- }
- Schema tagSchema = tagSchemaBuilder.build();
-
- Schema recordSchema = SchemaBuilder.struct()
- .name("org.apache.plc4x.kafka.schema.JobResult")
- .doc("PLC Job result. This contains all of the
received PLCValues as well as a received timestamp")
- .field(Constants.TAGS_CONFIG, tagSchema)
- .field(Constants.TIMESTAMP_CONFIG, Schema.INT64_SCHEMA)
- .field(Constants.EXPIRES_CONFIG,
Schema.OPTIONAL_INT64_SCHEMA)
- .build();
-
- // Build the struct itself.
- Struct tagStruct = new Struct(tagSchema);
- for (Map.Entry<String, Object> result :
results.entrySet()) {
- // Get tag-name and -value from the results.
- String tagName = result.getKey();
- Object tagValue = result.getValue();
-
- if (tagSchema.field(tagName).schema().type() ==
Schema.Type.ARRAY) {
- tagStruct.put(tagName, ((List)
tagValue).stream().map(p -> ((PlcValue)
p).getObject()).collect(Collectors.toList()));
- } else {
- tagStruct.put(tagName, tagValue);
- }
- }
+ // Get the schema for the given value type.
+ Schema valueSchema = getSchema(tagValue);
- Struct recordStruct = new Struct(recordSchema)
- .put(Constants.TAGS_CONFIG, tagStruct)
- .put(Constants.TIMESTAMP_CONFIG, timestamp);
-
- // Prepare the source-record element.
- SourceRecord sourceRecord = new SourceRecord(
- sourcePartition, sourceOffset,
- topic,
- KEY_SCHEMA, key,
- recordSchema, recordStruct
- );
-
- // Add the new source-record to the buffer.
- buffer.add(sourceRecord);
- } catch (Exception e) {
- log.error("Error while parsing returned values", e);
+ // Add the schema description for the current tag.
+ tagSchemaBuilder.field(tagName, valueSchema);
+ }
+ Schema tagSchema = tagSchemaBuilder.build();
+
+ Schema recordSchema = SchemaBuilder.struct()
+ .name("org.apache.plc4x.kafka.schema.JobResult")
+ .doc("PLC Job result. This contains all of the received
PLCValues as well as a received timestamp")
+ .field(Constants.TAGS_CONFIG, tagSchema)
+ .field(Constants.TIMESTAMP_CONFIG, Schema.INT64_SCHEMA)
+ .field(Constants.EXPIRES_CONFIG, Schema.OPTIONAL_INT64_SCHEMA)
+ .build();
+
+ // Build the struct itself.
+ Struct tagStruct = new Struct(tagSchema);
+ for (Map.Entry<String, Object> result : results.entrySet()) {
+ // Get tag-name and -value from the results.
+ String tagName = result.getKey();
+ Object tagValue = result.getValue();
+
+ if (tagSchema.field(tagName).schema().type() ==
Schema.Type.ARRAY) {
+ tagStruct.put(tagName, ((List) tagValue).stream().map(p ->
((PlcValue) p).getObject()).collect(Collectors.toList()));
+ } else {
+ tagStruct.put(tagName, tagValue);
}
- }, triggerCollector);
- scraper.start();
- triggerCollector.start();
- } catch (ScraperException e) {
- log.error("Error starting the scraper", e);
+ }
+ Struct recordStruct = new Struct(recordSchema)
+ .put(Constants.TAGS_CONFIG, tagStruct)
+ .put(Constants.TIMESTAMP_CONFIG, timestamp);
+
+ // Prepare the source-record element.
+ SourceRecord sourceRecord = new SourceRecord(
+ sourcePartition, sourceOffset,
+ topic,
+ KEY_SCHEMA, key,
+ recordSchema, recordStruct
+ );
+
+ // Add the new source-record to the buffer.
+ buffer.add(sourceRecord);
+ } catch (Exception e) {
+ log.error("Error while parsing returned values", e);
}
}
@Override
public void stop() {
synchronized (this) {
- scraper.stop();
+ if (eventPump != null) {
+ eventPump.close();
+ }
notifyAll(); // wake up thread waiting in awaitFetch
}
}
diff --git
a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SinkTaskTest.java
b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SinkTaskTest.java
index 27b20a34..e1c80d7d 100644
---
a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SinkTaskTest.java
+++
b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SinkTaskTest.java
@@ -79,7 +79,7 @@ public class SinkTaskTest {
assertEquals("machineSinkA",
config.get(0).get(Constants.TOPIC_CONFIG));
assertEquals("5", config.get(0).get(Constants.RETRIES_CONFIG));
assertEquals("5000", config.get(0).get(Constants.TIMEOUT_CONFIG));
-
assertEquals("running#RANDOM/Temporary:Boolean|conveyorEntry#RANDOM/Temporary:Boolean|load#RANDOM/Temporary:Boolean|unload#RANDOM/Temporary:Boolean|transferLeft#RANDOM/Temporary:Boolean|transferRight#RANDOM/Temporary:Boolean|conveyorLeft#RANDOM/Temporary:Boolean|conveyorRight#RANDOM/Temporary:Boolean|numLargeBoxes#STATE/Temporary:Integer|numSmallBoxes#RANDOM/Temporary:Integer",
+
assertEquals("running#RANDOM/Temporary:BOOL|conveyorEntry#RANDOM/Temporary:BOOL|load#RANDOM/Temporary:BOOL|unload#RANDOM/Temporary:BOOL|transferLeft#RANDOM/Temporary:BOOL|transferRight#RANDOM/Temporary:BOOL|conveyorLeft#RANDOM/Temporary:BOOL|conveyorRight#RANDOM/Temporary:BOOL|numLargeBoxes#STATE/Temporary:Integer|numSmallBoxes#RANDOM/Temporary:DINT",
config.get(0).get(Constants.QUERIES_CONFIG));
}
diff --git
a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SourceTaskTest.java
b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SourceTaskTest.java
index 2fdfe23b..dae0111c 100644
---
a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SourceTaskTest.java
+++
b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/config/SourceTaskTest.java
@@ -20,6 +20,7 @@ package org.apache.plc4x.kafka.config;
import static org.junit.jupiter.api.Assertions.*;
+import org.apache.kafka.connect.source.SourceRecord;
import org.apache.plc4x.kafka.Plc4xSourceConnector;
import org.apache.plc4x.kafka.Plc4xSourceTask;
@@ -69,7 +70,7 @@ public class SourceTaskTest {
assertEquals("simulated://127.0.0.1",
config.get(0).get(Constants.CONNECTION_STRING_CONFIG));
assertEquals("1000", config.get(0).get(Constants.BUFFER_SIZE_CONFIG));
assertEquals("5000",
config.get(0).get(Constants.KAFKA_POLL_RETURN_CONFIG));
-
assertEquals("simulateddashboard|machineData|1000|running#RANDOM/Temporary:Boolean|conveyorEntry#RANDOM/Temporary:Boolean|load#RANDOM/Temporary:Boolean|unload#RANDOM/Temporary:Boolean|transferLeft#RANDOM/Temporary:Boolean|transferRight#RANDOM/Temporary:Boolean|conveyorLeft#RANDOM/Temporary:Boolean|conveyorRight#RANDOM/Temporary:Boolean|numLargeBoxes#RANDOM/Temporary:Integer|numSmallBoxes#RANDOM/Temporary:Integer[2],simulatedheartbeat|simulatedheartbeat|500|active#RANDOM/Temporary
[...]
+
assertEquals("simulateddashboard|machineData|1000|running#RANDOM/Temporary:BOOL|conveyorEntry#RANDOM/Temporary:BOOL|load#RANDOM/Temporary:BOOL|unload#RANDOM/Temporary:BOOL|transferLeft#RANDOM/Temporary:BOOL|transferRight#RANDOM/Temporary:BOOL|conveyorLeft#RANDOM/Temporary:BOOL|conveyorRight#RANDOM/Temporary:BOOL|numLargeBoxes#RANDOM/Temporary:DINT|numSmallBoxes#RANDOM/Temporary:DINT[2],simulatedheartbeat|simulatedheartbeat|500|active#RANDOM/Temporary:DINT",
config.get(0).get(Cons [...]
}
@Test
@@ -85,8 +86,18 @@ public class SourceTaskTest {
sourceTask.start(taskConfig);
}
Thread.sleep(5000);
- for (Plc4xSourceTask sourceTask : sourceList) {
- assertNotNull(sourceTask.poll());
+ try {
+ for (Plc4xSourceTask sourceTask : sourceList) {
+ List<SourceRecord> records = sourceTask.poll();
+ assertNotNull(records);
+ // The jobs are configured to be read every 1000ms/500ms, so
after 5s there
+ // has to be data in the buffer - an empty result means
nothing was ever read.
+ assertFalse(records.isEmpty(), "expected the source task to
have collected records");
+ }
+ } finally {
+ for (Plc4xSourceTask sourceTask : sourceList) {
+ sourceTask.stop();
+ }
}
}
diff --git
a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/sink_task_no_error.properties
b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/sink_task_no_error.properties
index 68b31f37..ea26c4f1 100644
---
a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/sink_task_no_error.properties
+++
b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/sink_task_no_error.properties
@@ -26,24 +26,24 @@ sinks.machineA.timeout=5000
sinks.machineA.retries=5
sinks.machineA.tags=running,conveyorEntry,load,unload,transferLeft,transferRight,conveyorLeft,conveyorRight,numLargeBoxes,numSmallBoxes
-sinks.machineA.tags.running=RANDOM/Temporary:Boolean
-sinks.machineA.tags.conveyorEntry=RANDOM/Temporary:Boolean
-sinks.machineA.tags.load=RANDOM/Temporary:Boolean
-sinks.machineA.tags.unload=RANDOM/Temporary:Boolean
-sinks.machineA.tags.transferLeft=RANDOM/Temporary:Boolean
-sinks.machineA.tags.transferRight=RANDOM/Temporary:Boolean
-sinks.machineA.tags.conveyorLeft=RANDOM/Temporary:Boolean
-sinks.machineA.tags.conveyorRight=RANDOM/Temporary:Boolean
+sinks.machineA.tags.running=RANDOM/Temporary:BOOL
+sinks.machineA.tags.conveyorEntry=RANDOM/Temporary:BOOL
+sinks.machineA.tags.load=RANDOM/Temporary:BOOL
+sinks.machineA.tags.unload=RANDOM/Temporary:BOOL
+sinks.machineA.tags.transferLeft=RANDOM/Temporary:BOOL
+sinks.machineA.tags.transferRight=RANDOM/Temporary:BOOL
+sinks.machineA.tags.conveyorLeft=RANDOM/Temporary:BOOL
+sinks.machineA.tags.conveyorRight=RANDOM/Temporary:BOOL
sinks.machineA.tags.numLargeBoxes=STATE/Temporary:Integer
-sinks.machineA.tags.numSmallBoxes=RANDOM/Temporary:Integer
+sinks.machineA.tags.numSmallBoxes=RANDOM/Temporary:DINT
sinks.machineB.connectionString=simulated://127.0.0.1
sinks.machineB.topic=machineSinkB
sinks.machineB.timeout=5000
sinks.machineB.retries=5
sinks.machineB.tags=running,conveyorEntry
-sinks.machineB.tags.running=RANDOM/Temporary:Boolean
-sinks.machineB.tags.conveyorEntry=RANDOM/Temporary:Boolean
+sinks.machineB.tags.running=RANDOM/Temporary:BOOL
+sinks.machineB.tags.conveyorEntry=RANDOM/Temporary:BOOL
bootstrap.servers=127.0.0.1:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
diff --git
a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/source_task_no_error.properties
b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/source_task_no_error.properties
index adf250e1..a8f0a793 100644
---
a/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/source_task_no_error.properties
+++
b/plc4j/integrations/apache-kafka/src/test/java/org/apache/plc4x/kafka/properties/source_task_no_error.properties
@@ -31,20 +31,20 @@ jobs=simulateddashboard,simulatedheartbeat
jobs.simulateddashboard.interval=1000
jobs.simulateddashboard.tags=running,conveyorEntry,load,unload,transferLeft,transferRight,conveyorLeft,conveyorRight,numLargeBoxes,numSmallBoxes
-jobs.simulateddashboard.tags.running=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.conveyorEntry=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.load=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.unload=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.transferLeft=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.transferRight=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.conveyorLeft=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.conveyorRight=RANDOM/Temporary:Boolean
-jobs.simulateddashboard.tags.numLargeBoxes=RANDOM/Temporary:Integer
-jobs.simulateddashboard.tags.numSmallBoxes=RANDOM/Temporary:Integer[2]
+jobs.simulateddashboard.tags.running=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.conveyorEntry=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.load=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.unload=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.transferLeft=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.transferRight=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.conveyorLeft=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.conveyorRight=RANDOM/Temporary:BOOL
+jobs.simulateddashboard.tags.numLargeBoxes=RANDOM/Temporary:DINT
+jobs.simulateddashboard.tags.numSmallBoxes=RANDOM/Temporary:DINT[2]
jobs.simulatedheartbeat.interval=500
jobs.simulatedheartbeat.tags=active
-jobs.simulatedheartbeat.tags.active=RANDOM/Temporary:Integer
+jobs.simulatedheartbeat.tags.active=RANDOM/Temporary:DINT
bootstrap.servers=localhost:9092
#key.converter=org.apache.kafka.connect.json.JsonConverter
