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 b240c5503b Fix #1730: apply the camel.source.pollingConsumer* options
to the SEDA hand-off queue (#1870)
b240c5503b is described below
commit b240c5503b1925700c6d64ed213949d8701b315c
Author: Brian Hayden <[email protected]>
AuthorDate: Fri Nov 6 04:49:27 2026 -0500
Fix #1730: apply the camel.source.pollingConsumer* options to the SEDA
hand-off queue (#1870)
The source task receives exchanges through seda:end, but the
camel.source.pollingConsumer* options were only added to its URI as the
generic DefaultEndpoint polling-consumer options, which SedaEndpoint
ignores. The queue kept SEDA's defaults (size 1000, blockWhenFull=false),
so a burst failed with IllegalStateException: Queue full.
Pass the values as SEDA's own size, blockWhenFull and offerTimeout
options too. A negative pollingConsumerBlockTimeout is clamped to 0,
since a negative offerTimeout makes SEDA fall back to add() and throw
when full; the option is documented as "0 or negative means no timeout".
Behaviour change: with the default blockWhenFull=true the Camel consumer
thread now blocks while the queue is full, and the endpoint size takes
precedence over camel.component.seda.queueSize. A consumer blocked on a
full queue holds an in-flight exchange, so stopping the task can wait up
to camel.main.shutdownTimeout, longer than Kafka Connect's default
task.shutdown.graceful.timeout.ms of 5s. The user guide documents this.
Signed-off-by: Brian Hayden <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
core/pom.xml | 6 +
.../camel/kafkaconnector/CamelSourceTask.java | 8 +-
.../camel/kafkaconnector/CamelSourceTaskTest.java | 152 +++++++++++++++++++++
.../ROOT/pages/user-guide/basic-configuration.adoc | 11 ++
4 files changed, 176 insertions(+), 1 deletion(-)
diff --git a/core/pom.xml b/core/pom.xml
index 5f6f2706b3..e13b342070 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -125,6 +125,12 @@
<artifactId>camel-syslog</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility-version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-cassandraql</artifactId>
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
index 218bd17c1a..dbc330871f 100644
--- a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
+++ b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java
@@ -439,7 +439,13 @@ public class CamelSourceTask extends SourceTask {
private String getLocalUrlWithPollingOptions(long
pollingConsumerQueueSize, long pollingConsumerBlockTimeout, boolean
pollingConsumerBlockWhenFull) {
return LOCAL_URL + "?pollingConsumerQueueSize=" +
pollingConsumerQueueSize + "&pollingConsumerBlockTimeout=" +
pollingConsumerBlockTimeout
- + "&pollingConsumerBlockWhenFull=" +
pollingConsumerBlockWhenFull;
+ + "&pollingConsumerBlockWhenFull=" +
pollingConsumerBlockWhenFull
+ // SedaEndpoint polls its own queue and ignores the
pollingConsumer* options above, so the hand-off queue
+ // only gets these values through SEDA's own options. A
negative offerTimeout makes SEDA fall back to
+ // add(), which throws when the queue is full, so it is clamped
to 0 (block with no timeout), matching
+ // the documented "0 or negative then no timeout is in use".
+ + "&size=" + pollingConsumerQueueSize + "&blockWhenFull=" +
pollingConsumerBlockWhenFull
+ + "&offerTimeout=" + Math.max(0L, pollingConsumerBlockTimeout);
}
/**
diff --git
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
index 38e929adac..80c3f952ac 100644
---
a/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
+++
b/core/src/test/java/org/apache/camel/kafkaconnector/CamelSourceTaskTest.java
@@ -19,6 +19,7 @@ package org.apache.camel.kafkaconnector;
import java.io.ByteArrayInputStream;
import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -29,6 +30,8 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -36,6 +39,7 @@ import java.util.stream.IntStream;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.seda.SedaEndpoint;
import org.apache.camel.converter.stream.InputStreamCache;
import org.apache.camel.kafkaconnector.utils.StringJoinerAggregator;
import org.apache.camel.support.SynchronizationAdapter;
@@ -50,6 +54,7 @@ import org.junit.jupiter.api.Test;
import static org.apache.camel.util.CollectionHelper.mapOf;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -379,9 +384,156 @@ public class CamelSourceTaskTest {
assertTrue(e.getEndpointUri().contains("pollingConsumerBlockWhenFull=false"));
});
+ // SedaEndpoint ignores the generic pollingConsumer* options, the
values must reach its own options
+ SedaEndpoint seda = sedaEndpoint(sourceTask);
+ assertEquals(10, seda.getSize());
+ assertFalse(seda.isBlockWhenFull());
+ assertEquals(10L, seda.getOfferTimeout());
+
+ sourceTask.stop();
+ }
+
+ @Test
+ public void testSourcePollingConsumerDefaultOptions() {
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF,
DIRECT_URI);
+
+ CamelSourceTask sourceTask = new CamelSourceTask();
+ sourceTask.start(props);
+
+ SedaEndpoint seda = sedaEndpoint(sourceTask);
+
assertEquals(CamelSourceConnectorConfig.CAMEL_SOURCE_POLLING_CONSUMER_QUEUE_SIZE_DEFAULT.intValue(),
seda.getSize());
+ assertTrue(seda.isBlockWhenFull());
+ assertEquals(0L, seda.getOfferTimeout());
+
+ sourceTask.stop();
+ }
+
+ @Test
+ public void testSourcePollingConsumerNegativeBlockTimeoutMeansNoTimeout() {
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF,
DIRECT_URI);
+
props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_POLLING_CONSUMER_BLOCK_TIMEOUT_CONF,
"-1");
+
+ CamelSourceTask sourceTask = new CamelSourceTask();
+ sourceTask.start(props);
+
+ // a negative offerTimeout would make SEDA use add(), which throws
"Queue full" instead of blocking
+ assertEquals(0L, sedaEndpoint(sourceTask).getOfferTimeout());
+
+ sourceTask.stop();
+ }
+
+ @Test
+ public void testSourcePollingConsumerQueueSizeWinsOverComponentQueueSize()
{
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF,
DIRECT_URI);
+ // camel.source.pollingConsumerQueueSize is left at its default of 1000
+ props.put("camel.component.seda.queueSize", "7");
+
+ CamelSourceTask sourceTask = new CamelSourceTask();
+ sourceTask.start(props);
+
+ sendBatchOfRecords(sourceTask, 1);
+ SedaEndpoint seda = sedaEndpoint(sourceTask);
+ assertEquals(1, seda.getQueue().size());
+ assertEquals(999, seda.getQueue().remainingCapacity());
+
sourceTask.stop();
}
+ @Test
+ public void testSourcePollingConsumerBlocksWhenFull() throws Exception {
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF,
DIRECT_URI);
+
props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_POLLING_CONSUMER_QUEUE_SIZE_CONF,
"2");
+
+ CamelSourceTask sourceTask = new CamelSourceTask();
+ sourceTask.start(props);
+ SedaEndpoint seda = sedaEndpoint(sourceTask);
+
+ ExecutorService sender = Executors.newSingleThreadExecutor();
+ try {
+ AtomicInteger sent = new AtomicInteger();
+ Future<?> sending = sender.submit(() -> {
+ ProducerTemplate template =
sourceTask.getCms().getProducerTemplate();
+ for (int i = 0; i < 5; i++) {
+ template.sendBody(DIRECT_URI, "test" + i);
+ sent.incrementAndGet();
+ }
+ });
+
+ // the queue fills up and the third send stays blocked instead of
failing with "Queue full"
+
await().atMost(Duration.ofSeconds(10)).during(Duration.ofMillis(500))
+ .until(() -> sent.get() == 2 && seda.getQueue().size() ==
2 && !sending.isDone());
+
+ // draining the queue releases the producer
+ List<SourceRecord> records = new ArrayList<>();
+ await().atMost(Duration.ofSeconds(10)).until(() -> {
+ List<SourceRecord> polled = sourceTask.poll();
+ if (polled != null) {
+ records.addAll(polled);
+ }
+ return records.size() == 5;
+ });
+ sending.get(10, TimeUnit.SECONDS);
+ assertEquals(5, sent.get());
+ assertEquals(Arrays.asList("test0", "test1", "test2", "test3",
"test4"),
+
records.stream().map(SourceRecord::value).collect(Collectors.toList()));
+ } finally {
+ sender.shutdownNow();
+ sourceTask.stop();
+ }
+ }
+
+ @Test
+ public void testSourceStopWithProducerBlockedOnFullQueue() throws
Exception {
+ Map<String, String> props = new HashMap<>();
+ props.put(CamelSourceConnectorConfig.TOPIC_CONF, TOPIC_NAME);
+ props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_URL_CONF,
DIRECT_URI);
+
props.put(CamelSourceConnectorConfig.CAMEL_SOURCE_POLLING_CONSUMER_QUEUE_SIZE_CONF,
"1");
+ props.put("camel.main.shutdownTimeout", "1");
+
+ CamelSourceTask sourceTask = new CamelSourceTask();
+ sourceTask.start(props);
+ SedaEndpoint seda = sedaEndpoint(sourceTask);
+
+ ExecutorService sender = Executors.newSingleThreadExecutor();
+ boolean stopped = false;
+ try {
+ Future<?> sending = sender.submit(() ->
sendBatchOfRecords(sourceTask, 2));
+
await().atMost(Duration.ofSeconds(10)).during(Duration.ofMillis(300))
+ .until(() -> seda.getQueue().remainingCapacity() == 0 &&
!sending.isDone());
+
+ // the blocked send keeps an exchange in flight, so stopping waits
for camel.main.shutdownTimeout
+ long start = System.nanoTime();
+ stopped = true;
+ sourceTask.stop();
+ long stopMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime()
- start);
+
assertThat(stopMillis).isGreaterThanOrEqualTo(900L).isLessThan(10_000L);
+ // the forced stop does not interrupt the blocked thread: the
thread that owns it has to
+ assertFalse(sending.isDone());
+ } finally {
+ sender.shutdownNow();
+ if (!stopped) {
+ sourceTask.stop();
+ }
+ }
+ }
+
+ private static SedaEndpoint sedaEndpoint(CamelSourceTask sourceTask) {
+ return sourceTask.getCms().getCamelContext().getEndpoints().stream()
+ .filter(SedaEndpoint.class::isInstance)
+ .map(SedaEndpoint.class::cast)
+ .filter(e -> "end".equals(e.getName()))
+ .findFirst()
+ .orElseThrow();
+ }
+
@Test
public void testSourceUsingComponentProperties() {
Map<String, String> props = new HashMap<>();
diff --git a/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
b/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
index a9aacc2f7a..ed872625b3 100644
--- a/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
+++ b/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
@@ -88,6 +88,17 @@
transforms.toString.type=org.apache.camel.kafkaconnector.transforms.CamelTypeCon
transforms.toString.target.type=java.lang.String
----
+[[BasicConfiguration-SourceHandOffQueue]]
+== Source Hand-off Queue
+
+A source task receives messages from the Camel route through an internal SEDA
queue. The `camel.source.pollingConsumer*` options configure it:
+
+* `camel.source.pollingConsumerQueueSize` is the queue capacity. It takes
precedence over `camel.component.seda.queueSize`. A value of `0` or less does
not disable the queue: it falls back to `camel.component.seda.queueSize`, or to
1000.
+* With `camel.source.pollingConsumerBlockWhenFull=true` (the default), the
Camel consumer thread waits while the queue is full, until the task has polled
messages from it. With `false`, a message that arrives while the queue is full
fails with `IllegalStateException: Queue full`.
+* `camel.source.pollingConsumerBlockTimeout` limits that wait, in
milliseconds. When it expires, the message fails instead. `0` or a negative
value waits with no limit.
+
+A consumer thread that is waiting on a full queue holds an in-flight exchange.
When the task stops, Camel's graceful shutdown waits for it, up to
`camel.main.shutdownTimeout` (45 seconds by default), then forces the routes to
stop. Kafka Connect only waits `task.shutdown.graceful.timeout.ms` (5 seconds
by default) for the task to stop and logs an error when that runs out, so
consider setting `camel.main.shutdownTimeout` below it, or a non-zero
`camel.source.pollingConsumerBlockTimeout`. [...]
+
[[BasicConfiguration-MessageLevelHeaders]]
== Message Level Headers
