This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new e6433f423187 CAMEL-26151: camel-pg-replication-slot - keep the
generated exchange id and put the LSN in a header (#28921)
e6433f423187 is described below
commit e6433f4231872ef08bbc3fa61e63407403b392e3
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 2 20:46:40 2026 +0100
CAMEL-26151: camel-pg-replication-slot - keep the generated exchange id and
put the LSN in a header (#28921)
* 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.
* CAMEL-26151: camel-pg-replication-slot - escape attribute references …
…in the docs
Asciidoctor read ${...} in inline code as a document attribute reference,
which failed the
PR doc validation. Use the +...+ passthrough so the text shows as written.
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 990a5de7e5a8..e6138b4d018b 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
@@ -936,6 +936,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 3599c5ad2da7..00a87d5a1b14 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
@@ -3381,6 +3381,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) {
