This is an automated email from the ASF dual-hosted git repository. chrisdutz pushed a commit to branch chore/update-asf-yml in repository https://gitbox.apache.org/repos/asf/plc4x-extras.git
commit 9110c3a758f3d4c09b3e1cb43ae4e26c6abc5d40 Author: Christofer Dutz <[email protected]> AuthorDate: Thu Aug 13 17:58:25 2026 +0200 refactor: Updated the kafka and calcite modules to use the new event-pump instead of the now obsolete scraper. --- 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 +-- 18 files changed, 254 insertions(+), 194 deletions(-) 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
