bhayden53 opened a new pull request, #1870:
URL: https://github.com/apache/camel-kafka-connector/pull/1870
Fixes #1730
### Change
`CamelSourceTask` added the `camel.source.pollingConsumer*` options to the
`seda:end` URI only as the generic `DefaultEndpoint` polling-consumer options.
`SedaEndpoint` ignores them, so the hand-off queue kept SEDA's defaults (size
1000, `blockWhenFull=false`) and a burst failed with `IllegalStateException:
Queue full`.
- The same values are now also passed as SEDA's own `size`, `blockWhenFull`
and `offerTimeout` options.
- A negative `camel.source.pollingConsumerBlockTimeout` becomes
`offerTimeout=0`. A negative `offerTimeout` makes SEDA fall back to `add()`,
which throws when the queue is full. The option is documented as "0 or negative
then no timeout is in use".
- The user guide (`basic-configuration.adoc`) gets a short "Source Hand-off
Queue" section covering the three options and the stop behaviour.
### Behaviour change (for the release notes)
- With the documented default
`camel.source.pollingConsumerBlockWhenFull=true`, the Camel consumer thread now
blocks while the queue is full, until the task polls. Before, it failed with
`Queue full`.
- The endpoint size now takes precedence over
`camel.component.seda.queueSize`, including when
`camel.source.pollingConsumerQueueSize` is left at its default of 1000. Anyone
using `camel.component.seda.queueSize` as a workaround should move the value to
`camel.source.pollingConsumerQueueSize`.
- `camel.source.pollingConsumerQueueSize=0` (or less) does not disable the
queue, before or after this change: SEDA falls back to
`camel.component.seda.queueSize`, or 1000.
### Stop path
Blocking is new with this change, and so is this case. A consumer thread
blocked in `put()` holds an in-flight exchange, so stopping the task waits for
Camel's graceful shutdown, up to `camel.main.shutdownTimeout` (45s by default),
before Camel forces the routes to stop.
- Kafka Connect waits only `task.shutdown.graceful.timeout.ms` (5s by
default) for `stop()`, then logs `Graceful stop of task ... failed` and moves
on while the Camel context finishes stopping. The docs suggest keeping
`camel.main.shutdownTimeout` below it, or setting a non-zero
`camel.source.pollingConsumerBlockTimeout`.
- The forced stop does not interrupt the blocked thread; that is up to the
component that owns it. A message that was waiting is not delivered to Kafka,
the same as the messages already in the queue at stop. Sources that acknowledge
on commit redeliver it.
This PR documents the stop path and tests it; it does not change `stop()`.
Purging the queue during stop would release a blocked put sooner, at the cost
of dropping its message. I left that out to keep the change small, and can add
it if you prefer.
### Tests (`CamelSourceTaskTest`, Awaitility, no `Thread.sleep`)
- `testSourcePollingConsumerOptions` (extended): the SEDA endpoint's `size`,
`blockWhenFull` and `offerTimeout` match the configured values.
- `testSourcePollingConsumerDefaultOptions`: defaults give size 1000,
`blockWhenFull=true`, `offerTimeout=0`.
- `testSourcePollingConsumerNegativeBlockTimeoutMeansNoTimeout`: `-1`
becomes `offerTimeout=0`.
- `testSourcePollingConsumerQueueSizeWinsOverComponentQueueSize`: with
`camel.component.seda.queueSize=7` and the default `pollingConsumerQueueSize`,
the queue capacity is 1000.
- `testSourcePollingConsumerBlocksWhenFull`: with a queue of 2 and 5 sends,
the third send stays blocked and the queue holds 2. Polling releases it, and
all 5 records arrive in order.
- `testSourceStopWithProducerBlockedOnFullQueue`: with a send blocked on a
full queue and `camel.main.shutdownTimeout=1`, `stop()` returns after the 1s
graceful wait (asserted between 0.9s and 10s), and the send is still blocked
afterwards.
Awaitility is added to `core` as a test dependency (`${awaitility-version}`
from the Camel parent, as in the itests modules).
### Verification
- `mvn -pl core verify`: 147 tests, 0 failures (`CamelSourceTaskTest`: 32).
- Against `main` without the fix, 5 of the 6 tests above fail, for example
`testSourcePollingConsumerOptions: expected: <10> but was: <1000>`. The
negative-timeout test passes there too, since the old code never set
`offerTimeout`; it guards the clamp.
- `-Psourcecheck` is clean on `core`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
