bhayden53 commented on issue #1730: URL: https://github.com/apache/camel-kafka-connector/issues/1730#issuecomment-6536028344
We hit the same thing with the AWS SQS source connector on 4.14.0 (`IllegalStateException: Queue full` from `SedaProducer.addToQueue` -> `AbstractQueue.add`), and the same settings didn't help: `camel.component.seda.defaultBlockWhenFull=true` and `camel.source.pollingConsumer*`. Here's what we found, in case it helps with the RabbitMQ case. **Why neither set of options takes effect** 1. `camel.source.pollingConsumerQueueSize`, `pollingConsumerBlockWhenFull` and `pollingConsumerBlockTimeout` are added to the `seda:end` URI as the generic `DefaultEndpoint` polling-consumer options. But `SedaEndpoint` overrides `createPollingConsumer()` ([SedaEndpoint.java#L210](https://github.com/apache/camel/blob/camel-4.14.0/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaEndpoint.java#L210)) and polls its own queue directly, so those three options never reach the queue between the route and the task. That queue gets SEDA's own defaults: size 1000, and `blockWhenFull=false`, so `add()` throws when it's full. 2. `camel.component.seda.defaultBlockWhenFull` (and `defaultOfferTimeout`) are copied onto the endpoint when the endpoint is created ([SedaComponent.java#L270](https://github.com/apache/camel/blob/camel-4.14.0/components/camel-seda/src/main/java/org/apache/camel/component/seda/SedaComponent.java#L270)). `CamelSourceTask.start()` creates the `seda:end` endpoint ([CamelSourceTask.java#L188](https://github.com/apache/camel-kafka-connector/blob/main/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceTask.java#L188)) before `cms.start()` (L191), and `camel.component.*` properties are only applied during `cms.start()`. So the component ends up with the setting, but the endpoint has already been created without it. With DEBUG on `org.apache.camel.main` you can see `Configured property: camel.component.seda.defaultBlockWhenFull=true on bean: ...SedaComponent`, and the errors keep coming anyway. `camel.component.seda.queueSize` does take effect, because the queue itself is created lazily, later on. A larger queue only moves the limit, though. It doesn't add backpressure. **Fix we tested** Also pass the values as the SEDA endpoint's own options in `getLocalUrlWithPollingOptions`: ```diff private String getLocalUrlWithPollingOptions(long pollingConsumerQueueSize, long pollingConsumerBlockTimeout, boolean pollingConsumerBlockWhenFull) { return LOCAL_URL + "?pollingConsumerQueueSize=" + pollingConsumerQueueSize + "&pollingConsumerBlockTimeout=" + pollingConsumerBlockTimeout - + "&pollingConsumerBlockWhenFull=" + pollingConsumerBlockWhenFull; + + "&pollingConsumerBlockWhenFull=" + pollingConsumerBlockWhenFull + + "&size=" + pollingConsumerQueueSize + "&blockWhenFull=" + pollingConsumerBlockWhenFull + + "&offerTimeout=" + pollingConsumerBlockTimeout; } ``` With that patched into 4.14.0, a 20k-message SQS backlog went through with no `Queue full` errors. Messages received but not yet consumed stayed capped at about queue size × tasks, and `camel.source.pollingConsumerQueueSize` now actually sets the cap. The documented defaults (`blockWhenFull=true`, timeout 0) now mean the source consumer thread blocks until the task drains. One side effect: if the task is stopped while the consumer thread is blocked in `put()`, Camel's graceful shutdown waits for it, up to its default 45s. `camel.main.shutdownTimeout` bounds that, and in our test the forced stop didn't lose messages. Alternatively, a non-zero `pollingConsumerBlockTimeout` makes a blocked put give up after that many ms. Happy to open a PR with this if it looks like the right direction. -- 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]
