This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/main by this push:
new 4e2f6f6445 Fix #1803: hold back sink offsets until the aggregated
exchange is delivered (#1821)
4e2f6f6445 is described below
commit 4e2f6f64456ba111a4a0139c6300ef86c5de93e3
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Aug 25 06:38:24 2026 +0200
Fix #1803: hold back sink offsets until the aggregated exchange is
delivered (#1821)
put() sent each record synchronously and inspected the exchange, which is
sound
only while the route is fully synchronous. With aggregation configured, the
aggregate EIP completes the incoming exchange as soon as it has been merged
into
the buffer and delivers later on the aggregated exchange. So put() returned
clean, Kafka Connect committed the offset, and a failure of the aggregated
exchange was handled inside the route and dropped: no exception from put(),
no
reporter.report, no rewind. The whole batch was lost silently.
Track the records whose delivery is outstanding and implement preCommit to
commit only up to the oldest of them. DeliveryTrackingAggregationStrategy
wraps
the strategy named by camel.beans.aggregate so the records merged into an
aggregated exchange stay associated with it, and a completion on that
exchange
releases them once it has been delivered. A failed batch goes to the DLQ
when a
reporter is configured, and otherwise fails the task on the next call rather
than committing undelivered data.
Without aggregation the route is still synchronous, so records are released
as
put() returns and preCommit passes Kafka Connect's own offsets through
unchanged; testOffsetsAreNotHeldBackWithoutAggregation pins that.
Two things worth knowing for review. The strategy has to be decorated inside
configure() rather than in build(), because the bean named by
camel.beans.aggregate
is only bound once the context starts. And the accumulated records are
carried
from oldExchange on every call, because an aggregation strategy commonly
returns
newExchange, so the exchange carrying the batch changes identity as it
grows.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../apache/camel/kafkaconnector/CamelSinkTask.java | 68 ++++++++++++++++
.../DeliveryTrackingAggregationStrategy.java | 90 ++++++++++++++++++++++
.../kafkaconnector/SinkRecordDeliveryTracker.java | 90 ++++++++++++++++++++++
.../camel/kafkaconnector/SinkRecordReference.java | 54 +++++++++++++
.../utils/CamelKafkaConnectMain.java | 29 ++++++-
.../camel/kafkaconnector/CamelSinkTaskTest.java | 71 +++++++++++++++++
6 files changed, 401 insertions(+), 1 deletion(-)
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
index b36a738854..5ade2ccf72 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSinkTask.java
@@ -18,6 +18,7 @@ package org.apache.camel.kafkaconnector;
import java.util.Collection;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -31,6 +32,8 @@ import
org.apache.camel.kafkaconnector.utils.CamelKafkaConnectMain;
import org.apache.camel.kafkaconnector.utils.TaskHelper;
import org.apache.camel.support.DefaultExchange;
import org.apache.camel.util.StringHelper;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.errors.ConnectException;
@@ -48,6 +51,9 @@ public class CamelSinkTask extends SinkTask {
public static final String HEADER_CAMEL_PREFIX = "CamelHeader.";
public static final String PROPERTY_CAMEL_PREFIX = "CamelProperty.";
+ /** Set on each exchange so an aggregated delivery can be traced back to
the records it carries. */
+ static final String RECORD_REFERENCE_PROPERTY =
"CamelKafkaConnectorSinkRecord";
+
private static final String CAMEL_SINK_ENDPOINT_PROPERTIES_PREFIX =
"camel.sink.endpoint.";
private static final String CAMEL_SINK_PATH_PROPERTIES_PREFIX =
"camel.sink.path.";
@@ -56,6 +62,9 @@ public class CamelSinkTask extends SinkTask {
private static final String LOCAL_URL = "direct:start";
private static final String DEFAULT_KAMELET_CKC_SINK = "kamelet:ckcSink";
private ErrantRecordReporter reporter;
+ private final SinkRecordDeliveryTracker deliveryTracker = new
SinkRecordDeliveryTracker();
+ private volatile Throwable aggregatedDeliveryFailure;
+ private boolean aggregationEnabled;
private CamelKafkaConnectMain cms;
private ProducerTemplate producer;
@@ -98,6 +107,7 @@ public class CamelSinkTask extends SinkTask {
final String marshaller =
config.getString(CamelSinkConnectorConfig.CAMEL_SINK_MARSHAL_CONF);
final String unmarshaller =
config.getString(CamelSinkConnectorConfig.CAMEL_SINK_UNMARSHAL_CONF);
final int size =
config.getInt(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_SIZE_CONF);
+ aggregationEnabled =
config.getString(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_CONF) !=
null;
final long timeout =
config.getLong(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_TIMEOUT_CONF);
final int maxRedeliveries =
config.getInt(CamelSinkConnectorConfig.CAMEL_CONNECTOR_ERROR_HANDLER_MAXIMUM_REDELIVERIES_CONF);
final long redeliveryDelay =
config.getLong(CamelSinkConnectorConfig.CAMEL_CONNECTOR_ERROR_HANDLER_REDELIVERY_DELAY_CONF);
@@ -139,6 +149,7 @@ public class CamelSinkTask extends SinkTask {
.withProperties(actualProps)
.withUnmarshallDataFormat(unmarshaller)
.withMarshallDataFormat(marshaller)
+ .withAggregationStrategyDecorator(configured -> new
DeliveryTrackingAggregationStrategy(configured, this::onAggregatedExchangeDone))
.withAggregationSize(size)
.withAggregationTimeout(timeout)
.withErrorHandler(errorHandler)
@@ -191,10 +202,16 @@ public class CamelSinkTask extends SinkTask {
@Override
public void put(Collection<SinkRecord> sinkRecords) {
+ failIfAnAggregatedDeliveryFailed();
+
for (SinkRecord record : sinkRecords) {
TaskHelper.logRecordContent(LOG, loggingLevel, record);
+ SinkRecordReference reference = new SinkRecordReference(record);
+ deliveryTracker.inFlight(reference.topicPartition(),
reference.offset());
+
Exchange exchange = new
DefaultExchange(producer.getCamelContext());
+ exchange.setProperty(RECORD_REFERENCE_PROPERTY, reference);
exchange.getMessage().setBody(record.value());
exchange.getMessage().setHeader(KAFKA_RECORD_KEY_HEADER,
record.key());
@@ -214,6 +231,9 @@ public class CamelSinkTask extends SinkTask {
producer.send(localEndpoint, exchange);
if (exchange.isFailed()) {
+ // the record never reached the endpoint, so its offset must
not be held back on its behalf
+ deliveryTracker.delivered(reference.topicPartition(),
reference.offset());
+
if (reporter == null) {
LOG.warn("A delivery has failed and the error reporting is
NOT enabled. Records may be lost or ignored");
throw new ConnectException("Exchange delivery has
failed!", exchange.getException());
@@ -221,10 +241,58 @@ public class CamelSinkTask extends SinkTask {
LOG.warn("A delivery has failed and the error reporting is
enabled. Sending record to the DLQ");
reporter.report(record, exchange.getException());
+ } else if (!aggregationEnabled) {
+ // without aggregation the route is synchronous, so returning
here means the record was delivered
+ deliveryTracker.delivered(reference.topicPartition(),
reference.offset());
}
}
}
+ /**
+ * Holds back the offsets of records whose delivery has not completed.
Without aggregation every record is
+ * delivered by the time {@link #put} returns, so this is the set Kafka
Connect proposed. With aggregation the
+ * data of a record sits in the aggregation buffer after put() returns,
and its offset is only released once the
+ * aggregated exchange carrying it has completed.
+ */
+ @Override
+ public Map<TopicPartition, OffsetAndMetadata>
preCommit(Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
+ failIfAnAggregatedDeliveryFailed();
+ return deliveryTracker.safeOffsets(currentOffsets);
+ }
+
+ @Override
+ public void close(Collection<TopicPartition> partitions) {
+ deliveryTracker.forget(partitions);
+ super.close(partitions);
+ }
+
+ private void failIfAnAggregatedDeliveryFailed() {
+ Throwable failure = aggregatedDeliveryFailure;
+ if (failure != null) {
+ aggregatedDeliveryFailure = null;
+ throw new ConnectException("Delivery of an aggregated exchange has
failed!", failure);
+ }
+ }
+
+ /**
+ * Called when an aggregated exchange has completed, for every record that
went into it.
+ */
+ private void onAggregatedExchangeDone(List<SinkRecordReference>
references, Exchange exchange) {
+ if (exchange.isFailed()) {
+ if (reporter != null) {
+ LOG.warn("Delivery of an aggregated exchange failed and error
reporting is enabled. Sending {} record(s) to the DLQ",
+ references.size());
+ references.forEach(reference ->
reporter.report(reference.record(), exchange.getException()));
+ } else {
+ LOG.error("Delivery of an aggregated exchange carrying {}
record(s) failed and error reporting is NOT enabled",
+ references.size(), exchange.getException());
+ aggregatedDeliveryFailure = exchange.getException();
+ }
+ }
+
+ references.forEach(reference ->
deliveryTracker.delivered(reference.topicPartition(), reference.offset()));
+ }
+
@Override
public void stop() {
LOG.info("Stopping CamelSinkTask connector task");
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/DeliveryTrackingAggregationStrategy.java
b/core/src/main/java/org/apache/camel/kafkaconnector/DeliveryTrackingAggregationStrategy.java
new file mode 100644
index 0000000000..09903ae27c
--- /dev/null
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/DeliveryTrackingAggregationStrategy.java
@@ -0,0 +1,90 @@
+/*
+ * 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.kafkaconnector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.BiConsumer;
+
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.Exchange;
+import org.apache.camel.support.SynchronizationAdapter;
+
+/**
+ * Wraps the aggregation strategy named by the connector configuration,
keeping each record associated with the
+ * aggregated exchange it was merged into.
+ *
+ * The aggregate EIP completes an incoming exchange as soon as it has been
merged into the buffer, while the data is
+ * delivered later on the aggregated exchange. Without this, a record's offset
would be committed while its data was
+ * still in the aggregation buffer, and a failure of the aggregated exchange
would never reach the task.
+ */
+class DeliveryTrackingAggregationStrategy implements AggregationStrategy {
+
+ static final String AGGREGATED_RECORDS_PROPERTY =
"CamelKafkaConnectorAggregatedRecords";
+ static final String COMPLETION_REGISTERED_PROPERTY =
"CamelKafkaConnectorAggregatedCompletionRegistered";
+
+ private final AggregationStrategy delegate;
+ private final BiConsumer<List<SinkRecordReference>, Exchange>
onAggregatedExchangeDone;
+
+ DeliveryTrackingAggregationStrategy(AggregationStrategy delegate,
+ BiConsumer<List<SinkRecordReference>,
Exchange> onAggregatedExchangeDone) {
+ this.delegate = delegate;
+ this.onAggregatedExchangeDone = onAggregatedExchangeDone;
+ }
+
+ @Override
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ SinkRecordReference incoming = newExchange == null
+ ? null
+ :
newExchange.getProperty(CamelSinkTask.RECORD_REFERENCE_PROPERTY,
SinkRecordReference.class);
+
+ // An aggregation strategy commonly returns newExchange, so the
exchange carrying the batch changes identity on
+ // every call. Carry the accumulated records across so the list always
describes the whole batch.
+ List<SinkRecordReference> records = oldExchange == null ? new
ArrayList<>() : aggregatedRecords(oldExchange);
+
+ Exchange aggregated = delegate.aggregate(oldExchange, newExchange);
+
+ if (aggregated == null) {
+ return null;
+ }
+
+ if (incoming != null) {
+ records.add(incoming);
+ }
+ aggregated.setProperty(AGGREGATED_RECORDS_PROPERTY, records);
+
+ // Only the exchange that ends up leaving the aggregator completes, so
registering on each candidate is safe:
+ // the ones that are superseded never fire, and the one that does
holds the whole batch.
+ if (aggregated.getProperty(COMPLETION_REGISTERED_PROPERTY) == null) {
+ aggregated.setProperty(COMPLETION_REGISTERED_PROPERTY,
Boolean.TRUE);
+ aggregated.getExchangeExtension().addOnCompletion(new
SynchronizationAdapter() {
+ @Override
+ public void onDone(Exchange exchange) {
+ onAggregatedExchangeDone.accept(new ArrayList<>(records),
exchange);
+ }
+ });
+ }
+
+ return aggregated;
+ }
+
+ @SuppressWarnings("unchecked")
+ private List<SinkRecordReference> aggregatedRecords(Exchange exchange) {
+ List<SinkRecordReference> records =
exchange.getProperty(AGGREGATED_RECORDS_PROPERTY, List.class);
+ return records == null ? new ArrayList<>() : records;
+ }
+}
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/SinkRecordDeliveryTracker.java
b/core/src/main/java/org/apache/camel/kafkaconnector/SinkRecordDeliveryTracker.java
new file mode 100644
index 0000000000..d4927c2f24
--- /dev/null
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/SinkRecordDeliveryTracker.java
@@ -0,0 +1,90 @@
+/*
+ * 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.kafkaconnector;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NavigableSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListSet;
+
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+
+/**
+ * Tracks which sink records have been handed to the Camel route but not yet
delivered, so that
+ * {@link org.apache.kafka.connect.sink.SinkTask#preCommit} can hold back the
offsets of records whose delivery has
+ * not completed.
+ *
+ * A record is in flight from the moment it is sent into the route until the
exchange carrying it - which, when
+ * aggregation is configured, is the aggregated exchange rather than the one
the record entered on - has completed.
+ */
+class SinkRecordDeliveryTracker {
+
+ private final Map<TopicPartition, NavigableSet<Long>> inFlight = new
ConcurrentHashMap<>();
+
+ void inFlight(TopicPartition partition, long offset) {
+ inFlight.computeIfAbsent(partition, p -> new
ConcurrentSkipListSet<>()).add(offset);
+ }
+
+ void delivered(TopicPartition partition, long offset) {
+ NavigableSet<Long> offsets = inFlight.get(partition);
+ if (offsets != null) {
+ offsets.remove(offset);
+ }
+ }
+
+ /**
+ * Offsets safe to commit: for a partition with records still in flight,
everything strictly before the oldest of
+ * them; otherwise whatever Kafka Connect proposed.
+ */
+ Map<TopicPartition, OffsetAndMetadata> safeOffsets(Map<TopicPartition,
OffsetAndMetadata> currentOffsets) {
+ Map<TopicPartition, OffsetAndMetadata> safe = new
HashMap<>(currentOffsets.size());
+
+ for (Map.Entry<TopicPartition, OffsetAndMetadata> entry :
currentOffsets.entrySet()) {
+ NavigableSet<Long> offsets = inFlight.get(entry.getKey());
+ Long oldestInFlight = offsets == null || offsets.isEmpty() ? null
: offsets.first();
+
+ if (oldestInFlight == null || oldestInFlight >=
entry.getValue().offset()) {
+ safe.put(entry.getKey(), entry.getValue());
+ } else {
+ // commit up to, but not including, the oldest record still
awaiting delivery
+ safe.put(entry.getKey(), new
OffsetAndMetadata(oldestInFlight));
+ }
+ }
+
+ return safe;
+ }
+
+ /**
+ * Drops any state for partitions that are no longer assigned to this task.
+ */
+ void forget(Collection<TopicPartition> partitions) {
+ partitions.forEach(inFlight::remove);
+ }
+
+ void clear() {
+ inFlight.clear();
+ }
+
+ // visible for testing
+ int inFlightCount(TopicPartition partition) {
+ NavigableSet<Long> offsets = inFlight.get(partition);
+ return offsets == null ? 0 : offsets.size();
+ }
+}
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/SinkRecordReference.java
b/core/src/main/java/org/apache/camel/kafkaconnector/SinkRecordReference.java
new file mode 100644
index 0000000000..dd6a963b59
--- /dev/null
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/SinkRecordReference.java
@@ -0,0 +1,54 @@
+/*
+ * 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.kafkaconnector;
+
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.connect.sink.SinkRecord;
+
+/**
+ * Identifies a sink record while its delivery is outstanding, so its offset
can be held back until the exchange
+ * carrying it has completed. The record itself is kept so a failed delivery
can still be routed to the DLQ.
+ */
+final class SinkRecordReference {
+
+ private final TopicPartition topicPartition;
+ private final long offset;
+ private final SinkRecord record;
+
+ SinkRecordReference(SinkRecord record) {
+ this.topicPartition = new TopicPartition(record.topic(),
record.kafkaPartition());
+ this.offset = record.kafkaOffset();
+ this.record = record;
+ }
+
+ TopicPartition topicPartition() {
+ return topicPartition;
+ }
+
+ long offset() {
+ return offset;
+ }
+
+ SinkRecord record() {
+ return record;
+ }
+
+ @Override
+ public String toString() {
+ return topicPartition + "@" + offset;
+ }
+}
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
b/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
index 9d060769f0..87d2a825b3 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/utils/CamelKafkaConnectMain.java
@@ -20,8 +20,10 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.function.UnaryOperator;
import java.util.stream.Collectors;
+import org.apache.camel.AggregationStrategy;
import org.apache.camel.CamelContext;
import org.apache.camel.ConsumerTemplate;
import org.apache.camel.ProducerTemplate;
@@ -47,6 +49,7 @@ public class CamelKafkaConnectMain extends SimpleMain {
public static final String KAMELET_AGGREGATORL_TEMPLATE_PARAMETERS_PREFIX
= "camel.kamelet.ckcAggregator.";
public static final String KAMELET_IDEMPOTENT_TEMPLATE_PARAMETERS_PREFIX =
"camel.kamelet.ckcIdempotent.";
public static final String KAMELET_REMOVEHEADER_TEMPLATE_PARAMETERS_PREFIX
= "camel.kamelet.ckcRemoveHeader.";
+ public static final String CAMEL_CONNECTOR_TRACKED_AGGREGATE_NAME =
"ckcTrackedAggregate";
private static final Logger LOG =
LoggerFactory.getLogger(CamelKafkaConnectMain.class);
@@ -118,6 +121,7 @@ public class CamelKafkaConnectMain extends SimpleMain {
private int idempotentRepositoryKafkaPollDuration;
private String headersExcludePattern;
private boolean removeHeadersFirst;
+ private UnaryOperator<AggregationStrategy>
aggregationStrategyDecorator;
private boolean dumpRoutes = true;
public Builder(String from, String to) {
@@ -232,6 +236,16 @@ public class CamelKafkaConnectMain extends SimpleMain {
return this;
}
+ /**
+ * Wraps the aggregation strategy the configuration supplies, when
aggregation is in use. The sink task uses
+ * this to keep track of which records a given aggregated exchange
carries, since the aggregate EIP completes
+ * an incoming exchange before its data has been delivered.
+ */
+ public Builder
withAggregationStrategyDecorator(UnaryOperator<AggregationStrategy>
aggregationStrategyDecorator) {
+ this.aggregationStrategyDecorator = aggregationStrategyDecorator;
+ return this;
+ }
+
public Builder withDumpRoutes(boolean dumpRoutes) {
this.dumpRoutes = dumpRoutes;
return this;
@@ -320,6 +334,13 @@ public class CamelKafkaConnectMain extends SimpleMain {
camelMain.getCamelContext().getRegistry().bind("ckcIdempotentRepository",
idempotentRepo);
}
+ // The bean named by camel.beans.aggregate is only bound once the
context starts, so the strategy is
+ // decorated inside configure() below; the template just needs to
be pointed at the decorated name here.
+ if (aggregationStrategyDecorator != null) {
+
camelProperties.put(KAMELET_AGGREGATORL_TEMPLATE_PARAMETERS_PREFIX +
"aggregationStrategy",
+ CAMEL_CONNECTOR_TRACKED_AGGREGATE_NAME);
+ }
+
//remove headers
if (!ObjectHelper.isEmpty(headersExcludePattern)) {
camelProperties.put(KAMELET_REMOVEHEADER_TEMPLATE_PARAMETERS_PREFIX +
"headersExcludePattern", headersExcludePattern);
@@ -402,7 +423,13 @@ public class CamelKafkaConnectMain extends SimpleMain {
if (!ObjectHelper.isEmpty(unmarshallDataFormat)) {
rd = rd.kamelet("ckcUnMarshal");
}
- if (getContext().getRegistry().lookupByName("aggregate")
!= null) {
+ Object configuredAggregationStrategy =
+
getContext().getRegistry().lookupByName(CamelConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_NAME);
+ if (configuredAggregationStrategy != null) {
+ if (aggregationStrategyDecorator != null &&
configuredAggregationStrategy instanceof AggregationStrategy) {
+
getContext().getRegistry().bind(CAMEL_CONNECTOR_TRACKED_AGGREGATE_NAME,
+
aggregationStrategyDecorator.apply((AggregationStrategy)
configuredAggregationStrategy));
+ }
rd = rd.kamelet("ckcAggregator");
}
if (idempotencyEnabled) {
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
index eafcd45e15..26c76171ea 100644
--- a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
+++ b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSinkTaskTest.java
@@ -28,6 +28,8 @@ import java.util.Map;
import org.apache.camel.ConsumerTemplate;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaAndValue;
@@ -42,6 +44,7 @@ import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -1095,4 +1098,72 @@ class CamelSinkTaskTest {
Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
assertThat(exchange.getIn().getHeader(headerName)).isEqualTo(headerValue);
}
+
+ @Test
+ void testAggregatedRecordsOffsetsAreHeldBackUntilTheBatchIsDelivered() {
+ Map<String, String> props = new HashMap<>();
+ props.put(TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
+ props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_CONF,
"#class:org.apache.camel.kafkaconnector.utils.SampleAggregator");
+
props.put(CamelSinkConnectorConfig.CAMEL_CONNECTOR_AGGREGATE_SIZE_CONF, "5");
+ sinkTask.start(props);
+
+ TopicPartition partition = new TopicPartition(TOPIC_NAME, 1);
+
+ // four records: fewer than the aggregation size, so their data is
still sitting in the buffer
+ List<SinkRecord> firstBatch = new ArrayList<>();
+ for (int i = 0; i < 4; i++) {
+ firstBatch.add(new SinkRecord(TOPIC_NAME, 1, null, "test", null,
"camel" + i, 10 + i));
+ }
+ sinkTask.put(firstBatch);
+
+ Map<TopicPartition, OffsetAndMetadata> proposed =
Collections.singletonMap(partition, new OffsetAndMetadata(14));
+ Map<TopicPartition, OffsetAndMetadata> safe =
sinkTask.preCommit(proposed);
+ assertEquals(10, safe.get(partition).offset(),
+ "offsets must be held at the oldest record whose data is still
in the aggregation buffer");
+
+ // the fifth record completes the batch, so the aggregated exchange is
delivered
+ sinkTask.put(Collections.singletonList(new SinkRecord(TOPIC_NAME, 1,
null, "test", null, "camel4", 14)));
+
+ ConsumerTemplate consumer = sinkTask.getCms().getConsumerTemplate();
+ Exchange exchange = consumer.receive(SEDA_URI, RECEIVE_TIMEOUT);
+ assertNotNull(exchange, "the aggregated exchange should have been
delivered");
+
+ Map<TopicPartition, OffsetAndMetadata> afterDelivery =
Collections.singletonMap(partition, new OffsetAndMetadata(15));
+ Map<TopicPartition, OffsetAndMetadata> safeAfter =
awaitPreCommit(afterDelivery, 15);
+ assertEquals(15, safeAfter.get(partition).offset(),
+ "once the aggregated exchange has been delivered every offset
in it can be committed");
+ }
+
+ @Test
+ void testOffsetsAreNotHeldBackWithoutAggregation() {
+ Map<String, String> props = new HashMap<>();
+ props.put(TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSinkConnectorConfig.CAMEL_SINK_URL_CONF, SEDA_URI);
+ sinkTask.start(props);
+
+ TopicPartition partition = new TopicPartition(TOPIC_NAME, 1);
+ sinkTask.put(Collections.singletonList(new SinkRecord(TOPIC_NAME, 1,
null, "test", null, "camel", 10)));
+
+ Map<TopicPartition, OffsetAndMetadata> proposed =
Collections.singletonMap(partition, new OffsetAndMetadata(11));
+ assertEquals(11, sinkTask.preCommit(proposed).get(partition).offset(),
+ "without aggregation the route is synchronous, so nothing is
held back");
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata>
awaitPreCommit(Map<TopicPartition, OffsetAndMetadata> proposed, long expected) {
+ Map<TopicPartition, OffsetAndMetadata> safe = null;
+ for (int attempt = 0; attempt < 50; attempt++) {
+ safe = sinkTask.preCommit(proposed);
+ if (safe.values().iterator().next().offset() == expected) {
+ return safe;
+ }
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ break;
+ }
+ }
+ return safe;
+ }
}
