This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26151-pg-replication-slot-exchange-id in repository https://gitbox.apache.org/repos/asf/camel.git
commit 35f44b1facb98046fad2ae1e7c3de958bcd1ff21 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 17:37:29 2026 +0100 CAMEL-26151: camel-pg-replication-slot - keep the generated exchange id and put the LSN in a header The consumer set the exchange id to the last received LSN. Several messages share an LSN (the BEGIN and the first change of a transaction with test_decoding), and the value can be 0/0, so exchange ids repeated and anything keyed on the exchange id (idempotency, aggregation, tracing, the inflight registry) mixed up exchanges. The exchange keeps its generated id now, and the LSN is in the new CamelPgReplicationSlotLsn header. Upgrade guide entry added. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../catalog/components/pg-replication-slot.json | 5 +- .../docs/pg-replication-slot-component.adoc | 3 + .../pg/replication/slot/pg-replication-slot.json | 5 +- .../main/docs/pg-replication-slot-component.adoc | 3 + .../slot/PgReplicationSlotConstants.java | 35 +++++++++++ .../slot/PgReplicationSlotConsumer.java | 2 +- .../slot/PgReplicationSlotEndpoint.java | 6 +- .../PgReplicationSlotConsumerReconnectTest.java | 21 +++++-- .../slot/PgReplicationSlotConsumerTest.java | 68 ++++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +++ .../builder/endpoint/EndpointHeaderBuilders.java | 14 +++++ .../PgReplicationSlotEndpointBuilderFactory.java | 39 +++++++++++++ 12 files changed, 199 insertions(+), 10 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pg-replication-slot.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pg-replication-slot.json index db98358463cb..42d7c3358cbe 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pg-replication-slot.json +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/pg-replication-slot.json @@ -29,8 +29,11 @@ "healthCheckConsumerEnabled": { "index": 2, "kind": "property", "displayName": "Health Check Consumer Enabled", "group": "health", "label": "health", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Used for enabling or disabling all consumer based health checks from this component" }, "healthCheckProducerEnabled": { "index": 3, "kind": "property", "displayName": "Health Check Producer Enabled", "group": "health", "label": "health", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Used for enabling or disabling all producer based health checks from this component. Notice: Camel has by default disabled all producer based health-checks. You can turn on producer [...] }, + "headers": { + "CamelPgReplicationSlotLsn": { "index": 0, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The log sequence number (LSN) of the message, in the text form of PostgreSQL such as 0\/16B3748. Several messages can have the same LSN, for example the BEGIN and the first change of a transaction.", "constantName": "org.apache. [...] + }, "bodyTypes": { - "consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]", "description": "One message of the output plugin of the slot, as bytes: text for test_decoding, JSON for wal2json, the binary protocol for pgoutput; convert it to a String for the text plugins. The exchange id is the LSN of the message" } + "consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]", "description": "One message of the output plugin of the slot, as bytes: text for test_decoding, JSON for wal2json, the binary protocol for pgoutput; convert it to a String for the text plugins. The LSN of the message is in the CamelPgReplicationSlotLsn header" } }, "properties": { "slot": { "index": 0, "kind": "path", "displayName": "Slot", "group": "common", "label": "common", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Replication Slot name" }, diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/pg-replication-slot-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/pg-replication-slot-component.adoc index f91cf61a715f..1037af5e82a9 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/pg-replication-slot-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/pg-replication-slot-component.adoc @@ -55,6 +55,9 @@ The order of the messages is guaranteed, but the same message might come more th this component to sync data from PostgreSQL to another database, make sure your operations are idempotent (e.g., use `UPSERT` instead of `INSERT`, etc). This will make sure repeated messages won't affect your system negatively. +The log sequence number (LSN) of a message is in the `CamelPgReplicationSlotLsn` header. It is not unique: several +messages can have the same LSN, for example the `BEGIN` and the first change of a transaction with `test_decoding`. + == Examples .Example route diff --git a/components/camel-pg-replication-slot/src/generated/resources/META-INF/org/apache/camel/component/pg/replication/slot/pg-replication-slot.json b/components/camel-pg-replication-slot/src/generated/resources/META-INF/org/apache/camel/component/pg/replication/slot/pg-replication-slot.json index db98358463cb..42d7c3358cbe 100644 --- a/components/camel-pg-replication-slot/src/generated/resources/META-INF/org/apache/camel/component/pg/replication/slot/pg-replication-slot.json +++ b/components/camel-pg-replication-slot/src/generated/resources/META-INF/org/apache/camel/component/pg/replication/slot/pg-replication-slot.json @@ -29,8 +29,11 @@ "healthCheckConsumerEnabled": { "index": 2, "kind": "property", "displayName": "Health Check Consumer Enabled", "group": "health", "label": "health", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Used for enabling or disabling all consumer based health checks from this component" }, "healthCheckProducerEnabled": { "index": 3, "kind": "property", "displayName": "Health Check Producer Enabled", "group": "health", "label": "health", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Used for enabling or disabling all producer based health checks from this component. Notice: Camel has by default disabled all producer based health-checks. You can turn on producer [...] }, + "headers": { + "CamelPgReplicationSlotLsn": { "index": 0, "kind": "header", "displayName": "", "group": "consumer", "label": "consumer", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The log sequence number (LSN) of the message, in the text form of PostgreSQL such as 0\/16B3748. Several messages can have the same LSN, for example the BEGIN and the first change of a transaction.", "constantName": "org.apache. [...] + }, "bodyTypes": { - "consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]", "description": "One message of the output plugin of the slot, as bytes: text for test_decoding, JSON for wal2json, the binary protocol for pgoutput; convert it to a String for the text plugins. The exchange id is the LSN of the message" } + "consumerOut": { "carriers": [ "bytes" ], "javaType": "byte[]", "description": "One message of the output plugin of the slot, as bytes: text for test_decoding, JSON for wal2json, the binary protocol for pgoutput; convert it to a String for the text plugins. The LSN of the message is in the CamelPgReplicationSlotLsn header" } }, "properties": { "slot": { "index": 0, "kind": "path", "displayName": "Slot", "group": "common", "label": "common", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Replication Slot name" }, diff --git a/components/camel-pg-replication-slot/src/main/docs/pg-replication-slot-component.adoc b/components/camel-pg-replication-slot/src/main/docs/pg-replication-slot-component.adoc index f91cf61a715f..1037af5e82a9 100644 --- a/components/camel-pg-replication-slot/src/main/docs/pg-replication-slot-component.adoc +++ b/components/camel-pg-replication-slot/src/main/docs/pg-replication-slot-component.adoc @@ -55,6 +55,9 @@ The order of the messages is guaranteed, but the same message might come more th this component to sync data from PostgreSQL to another database, make sure your operations are idempotent (e.g., use `UPSERT` instead of `INSERT`, etc). This will make sure repeated messages won't affect your system negatively. +The log sequence number (LSN) of a message is in the `CamelPgReplicationSlotLsn` header. It is not unique: several +messages can have the same LSN, for example the `BEGIN` and the first change of a transaction with `test_decoding`. + == Examples .Example route diff --git a/components/camel-pg-replication-slot/src/main/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConstants.java b/components/camel-pg-replication-slot/src/main/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConstants.java new file mode 100644 index 000000000000..322cdd2732f2 --- /dev/null +++ b/components/camel-pg-replication-slot/src/main/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConstants.java @@ -0,0 +1,35 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.pg.replication.slot; + +import org.apache.camel.spi.Metadata; + +/** + * The headers of the pg-replication-slot component. + */ +public final class PgReplicationSlotConstants { + + @Metadata(label = "consumer", + description = "The log sequence number (LSN) of the message, in the text form of PostgreSQL such as 0/16B3748." + + " Several messages can have the same LSN, for example the BEGIN and the first change of a" + + " transaction.", + javaType = "String") + public static final String LSN = "CamelPgReplicationSlotLsn"; + + private PgReplicationSlotConstants() { + } +} diff --git a/components/camel-pg-replication-slot/src/main/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConsumer.java b/components/camel-pg-replication-slot/src/main/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConsumer.java index 50b9b263724b..b031dfac00a9 100644 --- a/components/camel-pg-replication-slot/src/main/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConsumer.java +++ b/components/camel-pg-replication-slot/src/main/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConsumer.java @@ -139,9 +139,9 @@ public class PgReplicationSlotConsumer extends ScheduledPollConsumer { } Exchange exchange = createExchange(true); - exchange.setExchangeId(stream.getLastReceiveLSN().asString()); Message message = exchange.getIn(); + message.setHeader(PgReplicationSlotConstants.LSN, stream.getLastReceiveLSN().asString()); message.setBody(this.payload); final long delay = this.endpoint.getStatusInterval(); diff --git a/components/camel-pg-replication-slot/src/main/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotEndpoint.java b/components/camel-pg-replication-slot/src/main/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotEndpoint.java index 76de2f10d477..ad1ecb8f5f1c 100644 --- a/components/camel-pg-replication-slot/src/main/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotEndpoint.java +++ b/components/camel-pg-replication-slot/src/main/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotEndpoint.java @@ -47,12 +47,12 @@ import org.slf4j.LoggerFactory; */ @UriEndpoint(firstVersion = "3.0.0", scheme = "pg-replication-slot", title = "PostgreSQL Replication Slot", syntax = "pg-replication-slot:host:port/database/slot:outputPlugin", - category = { Category.DATABASE }, consumerOnly = true) + category = { Category.DATABASE }, consumerOnly = true, headersClass = PgReplicationSlotConstants.class) @BodyTypes(consumerOut = @BodyType(value = Carrier.BYTES, javaType = "byte[]", description = "One message of the output plugin of the slot, as bytes: text for" + " test_decoding, JSON for wal2json, the binary protocol for pgoutput;" - + " convert it to a String for the text plugins. The exchange id is the LSN" - + " of the message")) + + " convert it to a String for the text plugins. The LSN of the message is in" + + " the CamelPgReplicationSlotLsn header")) public class PgReplicationSlotEndpoint extends ScheduledPollEndpoint { private static final Logger LOG = LoggerFactory.getLogger(PgReplicationSlotEndpoint.class); diff --git a/components/camel-pg-replication-slot/src/test/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConsumerReconnectTest.java b/components/camel-pg-replication-slot/src/test/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConsumerReconnectTest.java index 534c087071b5..6c6c429a0a75 100644 --- a/components/camel-pg-replication-slot/src/test/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConsumerReconnectTest.java +++ b/components/camel-pg-replication-slot/src/test/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConsumerReconnectTest.java @@ -186,7 +186,7 @@ class PgReplicationSlotConsumerReconnectTest { } } - private static PgReplicationSlotConsumer startRoute(DefaultCamelContext context, FakeServer server) throws Exception { + static PgReplicationSlotConsumer startRoute(DefaultCamelContext context, FakeServer server) throws Exception { PgReplicationSlotComponent component = new PgReplicationSlotComponent(); context.addComponent("pg-replication-slot", component); PgReplicationSlotEndpoint endpoint @@ -202,7 +202,11 @@ class PgReplicationSlotConsumerReconnectTest { @Override public void configure() { from(endpoint).routeId("slot") - .process(e -> server.deliver(new String(e.getIn().getBody(byte[].class), StandardCharsets.UTF_8))); + .process(e -> { + server.exchangeIds.add(e.getExchangeId()); + server.lsns.add(e.getIn().getHeader(PgReplicationSlotConstants.LSN, String.class)); + server.deliver(new String(e.getIn().getBody(byte[].class), StandardCharsets.UTF_8)); + }); } }); context.start(); @@ -220,7 +224,7 @@ class PgReplicationSlotConsumerReconnectTest { } } - private static void pollUntilReceived(PgReplicationSlotConsumer consumer, FakeServer server, int count) { + static void pollUntilReceived(PgReplicationSlotConsumer consumer, FakeServer server, int count) { for (int i = 0; i < MAX_POLLS && server.received.size() < count; i++) { try { consumer.poll(); @@ -240,6 +244,8 @@ class PgReplicationSlotConsumerReconnectTest { */ static final class FakeServer { final List<String> received = new CopyOnWriteArrayList<>(); + final List<String> exchangeIds = new CopyOnWriteArrayList<>(); + final List<String> lsns = new CopyOnWriteArrayList<>(); final List<String> pollFailures = new ArrayList<>(); private final List<String> slot = new ArrayList<>(); final List<FakeConnection> connections = new ArrayList<>(); @@ -248,6 +254,10 @@ class PgReplicationSlotConsumerReconnectTest { private boolean up = true; int failedDeliveries; SQLException nextReadError; + // the number of messages that were flushed: the LSN of a message is its position in the slot's history + int flushed; + // when set, the LSN of every message, as the BEGIN and the first change of a transaction have the same LSN + LogSequenceNumber sameLsn; FakeServer(Exception ioFailureCause) { this.ioFailureCause = ioFailureCause; @@ -386,7 +396,9 @@ class PgReplicationSlotConsumerReconnectTest { } return null; }); - when(stream.getLastReceiveLSN()).thenAnswer(i -> LogSequenceNumber.valueOf(sent)); + when(stream.getLastReceiveLSN()) + .thenAnswer( + i -> server.sameLsn != null ? server.sameLsn : LogSequenceNumber.valueOf(server.flushed + sent)); doAnswer(i -> { checkStreamOpen(); if (serverGone) { @@ -399,6 +411,7 @@ class PgReplicationSlotConsumerReconnectTest { for (int n = 0; n < sent; n++) { server.slot.remove(0); } + server.flushed += sent; sent = 0; return null; }).when(stream).forceUpdateStatus(); diff --git a/components/camel-pg-replication-slot/src/test/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConsumerTest.java b/components/camel-pg-replication-slot/src/test/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConsumerTest.java new file mode 100644 index 000000000000..0ee159d26505 --- /dev/null +++ b/components/camel-pg-replication-slot/src/test/java/org/apache/camel/component/pg/replication/slot/PgReplicationSlotConsumerTest.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.pg.replication.slot; + +import java.util.HashSet; +import java.util.List; + +import org.apache.camel.component.pg.replication.slot.PgReplicationSlotConsumerReconnectTest.FakeServer; +import org.apache.camel.impl.DefaultCamelContext; +import org.junit.jupiter.api.Test; +import org.postgresql.replication.LogSequenceNumber; + +import static org.apache.camel.component.pg.replication.slot.PgReplicationSlotConsumerReconnectTest.pollUntilReceived; +import static org.apache.camel.component.pg.replication.slot.PgReplicationSlotConsumerReconnectTest.startRoute; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * The exchanges of the consumer. The JDBC objects are the mocks of {@link PgReplicationSlotConsumerReconnectTest}. + */ +class PgReplicationSlotConsumerTest { + + @Test + void messagesWithTheSameLsnHaveTheirOwnExchangeId() throws Exception { + FakeServer server = new FakeServer(new IllegalStateException("not used")); + // test_decoding sends the BEGIN and the first change of a transaction with the same LSN + server.sameLsn = LogSequenceNumber.valueOf("0/16B3748"); + try (DefaultCamelContext context = new DefaultCamelContext()) { + PgReplicationSlotConsumer consumer = startRoute(context, server); + + server.publish("BEGIN 750"); + server.publish("table public.account: INSERT: id[integer]:1"); + pollUntilReceived(consumer, server, 2); + + assertEquals(2, new HashSet<>(server.exchangeIds).size(), + "the exchange ids are not unique: " + server.exchangeIds); + assertEquals(List.of("0/16B3748", "0/16B3748"), server.lsns, "the LSN header"); + } + } + + @Test + void theLsnOfTheMessageIsInAHeader() throws Exception { + FakeServer server = new FakeServer(new IllegalStateException("not used")); + try (DefaultCamelContext context = new DefaultCamelContext()) { + PgReplicationSlotConsumer consumer = startRoute(context, server); + + server.publish("first"); + server.publish("second"); + pollUntilReceived(consumer, server, 2); + + assertEquals(List.of("first", "second"), server.received); + assertEquals(List.of("0/1", "0/2"), server.lsns, "the LSN header"); + } + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 1bc188419d61..86098b5b3d2a 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -870,6 +870,14 @@ skipped too. Before, such a link was consumed like any other file, wherever it p the first time and at DEBUG on later polls. Links resolving inside the starting directory work as before. Set `jailStartingDirectory=false` to keep following links to other directories. +=== camel-pg-replication-slot - the LSN of a message is in the CamelPgReplicationSlotLsn header + +The consumer no longer sets the exchange id to the log sequence number (LSN) of the message. Several messages can have +the same LSN (for example the `BEGIN` and the first change of a transaction), so exchange ids repeated. The exchange +has a generated exchange id now, as with any other consumer, and the LSN is in the new `CamelPgReplicationSlotLsn` +header (a `String` such as `0/16B3748`). Use `${header.CamelPgReplicationSlotLsn}` instead of `${exchangeId}` to read +the LSN. + === camel-pqc - an unknown hybridKdfAlgorithm is rejected The hybrid KEM operations now fail with an `IllegalArgumentException` when `hybridKdfAlgorithm` is not one of diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointHeaderBuilders.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointHeaderBuilders.java index f33dda1a3c1e..dee8ead8a863 100644 --- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointHeaderBuilders.java +++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/EndpointHeaderBuilders.java @@ -3368,6 +3368,20 @@ public class EndpointHeaderBuilders { public static PdfEndpointBuilderFactory.PdfHeaderNameBuilder pdf() { return PdfEndpointBuilderFactory.PdfHeaderNameBuilder.INSTANCE; } + /** + * PostgreSQL Replication Slot (camel-pg-replication-slot) + * Poll for PostgreSQL Write-Ahead Log (WAL) records using Streaming + * Replication Slots. + * + * Category: database + * Since: 3.0 + * Maven coordinates: org.apache.camel:camel-pg-replication-slot + * + * @return the dsl builder for the headers' name. + */ + public static PgReplicationSlotEndpointBuilderFactory.PgReplicationSlotHeaderNameBuilder pgReplicationSlot() { + return PgReplicationSlotEndpointBuilderFactory.PgReplicationSlotHeaderNameBuilder.INSTANCE; + } /** * PostgreSQL Event (camel-pgevent) * Send and receive PostgreSQL events via LISTEN and NOTIFY commands. diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PgReplicationSlotEndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PgReplicationSlotEndpointBuilderFactory.java index aaa477d38f51..88dd73e49015 100644 --- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PgReplicationSlotEndpointBuilderFactory.java +++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/PgReplicationSlotEndpointBuilderFactory.java @@ -834,6 +834,20 @@ public interface PgReplicationSlotEndpointBuilderFactory { } public interface PgReplicationSlotBuilders { + /** + * PostgreSQL Replication Slot (camel-pg-replication-slot) + * Poll for PostgreSQL Write-Ahead Log (WAL) records using Streaming + * Replication Slots. + * + * Category: database + * Since: 3.0 + * Maven coordinates: org.apache.camel:camel-pg-replication-slot + * + * @return the dsl builder for the headers' name. + */ + default PgReplicationSlotHeaderNameBuilder pgReplicationSlot() { + return PgReplicationSlotHeaderNameBuilder.INSTANCE; + } /** * PostgreSQL Replication Slot (camel-pg-replication-slot) * Poll for PostgreSQL Write-Ahead Log (WAL) records using Streaming @@ -908,6 +922,31 @@ public interface PgReplicationSlotEndpointBuilderFactory { } } + /** + * The builder of headers' name for the PostgreSQL Replication Slot component. + */ + public static class PgReplicationSlotHeaderNameBuilder { + /** + * The internal instance of the builder used to access to all the + * methods representing the name of headers. + */ + public static final PgReplicationSlotHeaderNameBuilder INSTANCE = new PgReplicationSlotHeaderNameBuilder(); + + /** + * The log sequence number (LSN) of the message, in the text form of + * PostgreSQL such as 0/16B3748. Several messages can have the same LSN, + * for example the BEGIN and the first change of a transaction. + * + * The option is a: {@code String} type. + * + * Group: consumer + * + * @return the name of the header {@code PgReplicationSlotLsn}. + */ + public String pgReplicationSlotLsn() { + return "CamelPgReplicationSlotLsn"; + } + } static PgReplicationSlotEndpointBuilder endpointBuilder(String componentName, String path) { class PgReplicationSlotEndpointBuilderImpl extends AbstractEndpointBuilder implements PgReplicationSlotEndpointBuilder, AdvancedPgReplicationSlotEndpointBuilder { public PgReplicationSlotEndpointBuilderImpl(String path) {
