You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "allthingssecurity (via GitHub)" <gi...@apache.org> on 2026/10/16 00:05:30 UTC
[PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
allthingssecurity opened a new pull request, #27922:
URL: https://github.com/apache/camel/pull/27922
# Description
[CAMEL-25683](https://issues.apache.org/jira/browse/CAMEL-25683)
`EventHubsConsumer.processCommit` keeps, per partition, a checkpoint task that the scheduler runs `checkpointBatchTimeout` after the first event of a window, and checkpoints at once when `checkpointBatchSize` events have completed. When the batch size was reached it cancelled the scheduled future but left the task registered. Until the time that task had been scheduled for, the following events of the partition found it "not expired" and only set their event context on it, so nothing checkpointed them by time. If no event of the partition completed after that time (the end of a burst), those events stayed uncheckpointed until the next event, and a restart or the partition moving to another consumer delivered them again (up to `checkpointBatchSize - 1` per partition; with the defaults, a burst of 700 events within 5 seconds leaves events 501-700 uncheckpointed).
This change removes the task and its future from the two maps when the batch size is reached, after cancelling the future, so the next event of the partition starts a new timeout window, as the first event after start does. The size checkpoint, the timer task and the `CamelAzureEventHubsCheckpointUpdatedBy` header work as before. A partition can now get at most one more timer checkpoint per size checkpoint (one Blob write) than before.
A second, smaller cause of the same symptom is fixed with it: `EventHubsCheckpointUpdaterTask.run()` reset the per-partition counter (`processedEvents.set(0)`) in the completion callback of its checkpoint, on a reactor thread and without the consumer's lock. The counter is shared with the next task of the partition, so an event that completed while that checkpoint was being stored (and started the next window) was wiped from the count, and the next task then found 0 and skipped its checkpoint. The callback now resets the counter only if it still holds the value the task checkpointed (`compareAndSet`); otherwise the count stays a little high, which only makes the next size checkpoint come earlier.
No upgrade guide entry: no option or default changes, and the timer checkpoints at most once per window as the option documentation says ("Works jointly with checkpointBatchSize"); users only see the missing checkpoints made.
Tests:
- New `EventHubsConsumerCheckpointAfterBatchSizeTest` (3 tests, in the style of `EventHubsConsumerPerPartitionCheckpointTest`: mocked `EventContext`, `processCommit` called directly, but with a real single-thread scheduler; batch size 2, timeout 1000 ms): an event after a size checkpoint, and an event after two size checkpoints, must be checkpointed by the timeout when no other event follows; and an event that completes while the timeout checkpoint of the previous window is still being stored (its `Mono` completes when the test says so) must be checkpointed by the next timeout. Without the change each fails after 10 s with `Wanted but not invoked: eventContext.updateCheckpointAsync()` (the first two without the change in `processCommit`, the third without the change in `EventHubsCheckpointUpdaterTask`).
- camel-azure-eventhubs unit tests with the change: 33 run, 0 failures. The integration tests (Azure) were not run.
# Target
- [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch)
# Tracking
- [ ] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it).
(The ticket is drafted and not filed yet; CAMEL-25683 is a placeholder until it is.)
# Apache Camel coding standards and style
- [x] I checked that each commit in the pull request has a meaningful subject line and body.
- [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes.
(I built and tested the affected module, including the formatter and import-sort plugins. I did not run the full root build.)
# AI-assisted contributions
- [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used.
This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer.
_Claude Code on behalf of allthingssecurity_
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4291296662
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -40,14 +40,17 @@ public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger p
@Override
public void run() {
- if (processedEvents.get() > 0) {
- LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processedEvents.get());
+ final int processed = processedEvents.get();
+ if (processed > 0) {
+ LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processed);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
error -> LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error),
() -> {
LOG.debug("Checkpoint updated.");
- processedEvents.set(0);
+ // the counter is shared with the next task of the partition: an event that completed
+ // while this checkpoint was being stored is counted for that task, so keep it
+ processedEvents.compareAndSet(processed, 0);
Review Comment:
Thanks. You are right that the `compareAndSet` still wipes the next window when the count climbs back to exactly the checkpointed value after a size checkpoint during the store round trip. Your `getAndSet(0)` when the task runs, with the count added back in the error callback, closes that case. I would rather do that as a small follow-up with its own test (it changes when the counter is reset, and the failure path needs a test of its own), so this PR keeps the two fixes it has. If you prefer it here, I can add it.
_Claude Code on behalf of allthingssecurity_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4292538198
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -40,15 +40,19 @@ public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger p
@Override
public void run() {
- if (processedEvents.get() > 0) {
- LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processedEvents.get());
+ // take the count when the checkpoint is started: an event that completes while it is being stored is counted
+ // for the next window from 0, and a checkpoint by batch size in the meantime cannot be confused with this one
+ final int processed = processedEvents.getAndSet(0);
+ if (processed > 0) {
+ LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processed);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
- error -> LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error),
- () -> {
- LOG.debug("Checkpoint updated.");
- processedEvents.set(0);
- });
+ error -> {
+ LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error);
+ // these events are not checkpointed: count them again for the next checkpoint
+ processedEvents.addAndGet(processed);
Review Comment:
⚠️ **Concurrency note:** `addAndGet(processed)` runs on the reactor thread without the consumer `lock`, on the same `AtomicInteger` that `processCommit` increments under `lock`. If the checkpoint store fails *after* a new window has already started for the same partition (new events have pushed the counter above zero), the restored count inflates the new window — potentially triggering a premature size-checkpoint. The result is an extra Blob write, not a missed checkpoint, so the failure mode is benign. Worth a comment here explaining the deliberate tradeoff (over-count on store failure → early checkpoint) to distinguish it from the pre-existing symmetric gap in the size-checkpoint path (which currently loses the count on failure with no restore).
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4292682331
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -40,15 +40,19 @@ public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger p
@Override
public void run() {
- if (processedEvents.get() > 0) {
- LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processedEvents.get());
+ // take the count when the checkpoint is started: an event that completes while it is being stored is counted
+ // for the next window from 0, and a checkpoint by batch size in the meantime cannot be confused with this one
+ final int processed = processedEvents.getAndSet(0);
+ if (processed > 0) {
+ LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processed);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
- error -> LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error),
- () -> {
- LOG.debug("Checkpoint updated.");
- processedEvents.set(0);
- });
+ error -> {
+ LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error);
+ // these events are not checkpointed: count them again for the next checkpoint
+ processedEvents.addAndGet(processed);
Review Comment:
Agreed, that is the intended trade-off: after a failed store the restored count can only make the next size checkpoint come a little earlier (one extra checkpoint write), never skip one. The code comment at that line says the events are counted again for the next checkpoint; I can extend it with the "may over-count, never under-count" wording if a maintainer wants it.
_Claude Code on behalf of allthingssecurity_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4294662591
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java:
##########
@@ -246,13 +246,18 @@ private void processCommit(final Exchange exchange, final EventContext eventCont
try {
var cnt = processedEvents.incrementAndGet();
- if (cnt == getConfiguration().getCheckpointBatchSize()) {
+ // >= and not ==: a failed timeout checkpoint adds its count back, so the count can pass the batch size
+ if (cnt >= getConfiguration().getCheckpointBatchSize()) {
processedEvents.set(0);
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY, COMPLETED_BY_SIZE);
LOG.debug("eventhub consumer batch size of reached for partition {}", partitionId);
if (scheduledTask != null) {
scheduledTask.cancel(false);
}
+ // the next event starts a new timeout window: the cancelled task must not be reused until its
+ // scheduled time, else the events of the rest of that window are not checkpointed by the timeout
+ checkpointTasksByPartition.remove(partitionId);
+ scheduledTasksByPartition.remove(partitionId);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
error -> LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error),
Review Comment:
Thanks. I'd keep the batch-size path as it is in this PR, because adding the count back there would not change what gets checkpointed:
- A checkpoint stores the offset of the latest event, so the next checkpoint of the partition (by size or by timeout) also covers the events of a failed one. The count only decides *when* that next checkpoint happens; restoring it would just make the next size checkpoint come a little earlier.
- The case that is really left open is a failed size checkpoint followed by no further event: the timer task of that window was cancelled, so nothing checkpoints those events until the next event arrives (on a restart they are delivered again). Restoring the counter does not help there either, since no task is scheduled to look at it. That is the same on main today (the batch-size path only logs), so I've noted it as a separate follow-up rather than widening this PR: a failed size checkpoint should schedule a timeout checkpoint for the same events.
In the timer path the add-back matters because that task skips the checkpoint when its count is 0.
_Claude Code on behalf of allthingssecurity_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27922:
URL: https://github.com/apache/camel/pull/27922#issuecomment-6211122721
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4294618862
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java:
##########
@@ -246,13 +246,18 @@ private void processCommit(final Exchange exchange, final EventContext eventCont
try {
var cnt = processedEvents.incrementAndGet();
- if (cnt == getConfiguration().getCheckpointBatchSize()) {
+ // >= and not ==: a failed timeout checkpoint adds its count back, so the count can pass the batch size
+ if (cnt >= getConfiguration().getCheckpointBatchSize()) {
processedEvents.set(0);
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY, COMPLETED_BY_SIZE);
LOG.debug("eventhub consumer batch size of reached for partition {}", partitionId);
if (scheduledTask != null) {
scheduledTask.cancel(false);
}
+ // the next event starts a new timeout window: the cancelled task must not be reused until its
+ // scheduled time, else the events of the rest of that window are not checkpointed by the timeout
+ checkpointTasksByPartition.remove(partitionId);
+ scheduledTasksByPartition.remove(partitionId);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
error -> LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error),
Review Comment:
⚠️ **Minor inconsistency:** The timer path (`EventHubsCheckpointUpdaterTask`) restores the event count on checkpoint failure via `processedEvents.addAndGet(processed)`, so that those events aren't silently dropped from the counter. The batch-size path here only logs the error — if the checkpoint call fails, those `processed` events are lost from the counter and won't contribute to the next batch-size threshold. Consider adding the same recovery here:
```suggestion
error -> {
LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error);
processedEvents.addAndGet((int) getConfiguration().getCheckpointBatchSize());
},
```
(Or extract the error-recovery lambda into a shared helper to avoid duplication.)
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus merged PR #27922:
URL: https://github.com/apache/camel/pull/27922
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4291166694
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -40,14 +40,17 @@ public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger p
@Override
public void run() {
- if (processedEvents.get() > 0) {
- LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processedEvents.get());
+ final int processed = processedEvents.get();
+ if (processed > 0) {
+ LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processed);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
error -> LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error),
() -> {
LOG.debug("Checkpoint updated.");
- processedEvents.set(0);
+ // the counter is shared with the next task of the partition: an event that completed
+ // while this checkpoint was being stored is counted for that task, so keep it
+ processedEvents.compareAndSet(processed, 0);
Review Comment:
Nice catch on the shared counter. One remaining corner, not a blocker: if a full batch completes while this checkpoint is being stored (the size path sets the counter to 0) and the count then climbs back to exactly `processed`, the `compareAndSet` still succeeds and wipes those events from the next window. An alternative without that case is to take the count when the task runs (`int processed = processedEvents.getAndSet(0);`) and add it back in the error callback if the checkpoint fails. Happy to leave it as is if you think the window is too small to matter.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27922:
URL: https://github.com/apache/camel/pull/27922#issuecomment-6212133199
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `components/camel-azure/camel-azure-eventhubs`
---
<details><summary>:microscope: Scalpel shadow comparison — Scalpel: 8 of 706 tested, 27 compile-only — current: 8 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **8 affected modules** (current approach: 8).
**Skip-tests mode would test 8 modules** (1 direct + 7 downstream), **skip tests for 27** (generated code, meta-modules)
<details><summary>Modules Scalpel would test (8)</summary>
- `camel-azure-eventhubs` ← components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java, components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java, components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointAfterBatchSizeTest.java
- `camel-jbang-mcp` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-mcp` ← downstream of org.apache.camel:camel-jbang-core
- `camel-jbang-plugin-tui` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-validate` ← downstream of org.apache.camel:camel-yaml-dsl-validator
- `camel-launcher-container` ← downstream of org.apache.camel:camel-launcher
- `camel-yaml-dsl-validator` ← downstream of org.apache.camel:camel-catalog
- `camel-yaml-dsl-validator-maven-plugin` ← downstream of org.apache.camel:camel-yaml-dsl-validator
</details>
<details><summary>Modules with tests skipped (27)</summary>
- `apache-camel`
- `camel-allcomponents`
- `camel-catalog`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-componentdsl`
- `camel-endpointdsl`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-kamelet-main`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `coverage`
- `docs`
- `dummy-component`
</details>
> :information_source: Shadow mode — Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
<details><summary>All tested modules (35 modules, 5m 27s total)</summary>
**Total reactor time:** 5m 27s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: Launcher | 49.7s | SUCCESS |
| Camel :: JBang :: MCP | 41.1s | SUCCESS |
| Camel :: JBang :: Plugin :: Validate | 32.9s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 22.1s | SUCCESS |
| Camel :: Component DSL | 20.3s | SUCCESS |
| Camel :: YAML DSL :: Validator | 19.7s | SUCCESS |
| Camel :: Azure :: Event Hubs | 19.2s | SUCCESS |
| Camel :: YAML DSL | 18.4s | SUCCESS |
| Camel :: Docs | 14.6s | SUCCESS |
| Camel :: Kamelet Main | 13.3s | SUCCESS |
| Camel :: JBang :: Plugin :: Kubernetes | 12.3s | SUCCESS |
| Camel :: JBang :: Plugin :: Testing | 8.6s | SUCCESS |
| Camel :: YAML DSL :: Deserializers | 8.2s | SUCCESS |
| Camel :: Catalog :: Camel Route Parser | 7.3s | SUCCESS |
| Camel :: Catalog :: Camel Report Maven Plugin | 6.6s | SUCCESS |
| Camel :: YAML DSL :: Validator Maven Plugin | 6.4s | SUCCESS |
| Camel :: All Components Sync point | 5.2s | SUCCESS |
| Camel :: YAML DSL :: Maven Plugins | 3.3s | SUCCESS |
| Camel :: Catalog :: Maven | 3.1s | SUCCESS |
| Camel :: Catalog :: Suggest (deprecated) | 2.3s | SUCCESS |
| Camel :: JBang :: Plugin :: Edit | 1.8s | SUCCESS |
| Camel :: Assembly | 1.7s | SUCCESS |
| Camel :: JBang :: Integration tests | 1.5s | SUCCESS |
| Camel :: Coverage | 1.2s | SUCCESS |
| Camel :: Catalog :: Dummy Component | 1.1s | SUCCESS |
| Camel :: JBang :: Plugin :: Generate | 1.0s | SUCCESS |
| Camel :: JBang :: Main | 0.9s | SUCCESS |
| Camel :: Catalog :: Console | 0.9s | SUCCESS |
| Camel :: Endpoint DSL :: Support | 0.8s | SUCCESS |
| Camel :: Launcher :: Container | 0.7s | SUCCESS |
| Camel :: JBang :: Plugin :: MCP | 0.5s | SUCCESS |
| Camel :: Endpoint DSL | n/a | |
| Camel :: Integration Tests | n/a | |
| Camel :: JBang :: Core | n/a | |
| Camel :: JBang :: Plugin :: TUI | n/a | |
**Top 20 slowest modules:**
- `Camel :: Launcher` (49.7s)
- `Camel :: JBang :: MCP` (41.1s)
- `Camel :: JBang :: Plugin :: Validate` (32.9s)
- `Camel :: Catalog :: Camel Catalog` (22.1s)
- `Camel :: Component DSL` (20.3s)
- `Camel :: YAML DSL :: Validator` (19.7s)
- `Camel :: Azure :: Event Hubs` (19.2s)
- `Camel :: YAML DSL` (18.4s)
- `Camel :: Docs` (14.6s)
- `Camel :: Kamelet Main` (13.3s)
- `Camel :: JBang :: Plugin :: Kubernetes` (12.3s)
- `Camel :: JBang :: Plugin :: Testing` (8.6s)
- `Camel :: YAML DSL :: Deserializers` (8.2s)
- `Camel :: Catalog :: Camel Route Parser` (7.3s)
- `Camel :: Catalog :: Camel Report Maven Plugin` (6.6s)
- `Camel :: YAML DSL :: Validator Maven Plugin` (6.4s)
- `Camel :: All Components Sync point` (5.2s)
- `Camel :: YAML DSL :: Maven Plugins` (3.3s)
- `Camel :: Catalog :: Maven` (3.1s)
- `Camel :: Catalog :: Suggest (deprecated)` (2.3s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/38870502522)
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4292195477
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -40,14 +40,17 @@ public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger p
@Override
public void run() {
- if (processedEvents.get() > 0) {
- LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processedEvents.get());
+ final int processed = processedEvents.get();
+ if (processed > 0) {
+ LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processed);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
error -> LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error),
() -> {
LOG.debug("Checkpoint updated.");
- processedEvents.set(0);
+ // the counter is shared with the next task of the partition: an event that completed
+ // while this checkpoint was being stored is counted for that task, so keep it
+ processedEvents.compareAndSet(processed, 0);
Review Comment:
Update: I changed my mind and did it in this PR, in aaccfdf9fc45, as you suggested. I measured the window first instead of guessing: a deterministic simulation (5000 seeded runs, with some slow checkpoint-store answers) lost the last events of a partition in 31 runs through exactly this case with the `compareAndSet` version, so it is not too small to matter.
The timer task now takes the count when it starts the checkpoint (`getAndSet(0)`) and adds it back in the error callback if storing fails. New test `eventAfterBatchSizeCheckpointDuringTimeoutCheckpointIsCheckpointed` (batch size 3): while the timeout checkpoint of the first event is stored, three more events complete, the third after a size checkpoint, and the last event must still be checkpointed. It fails with the previous commit (`Wanted but not invoked` after 10 s) and passes now. camel-azure-eventhubs tests: 34 run, 0 failures. The description is updated.
_Claude Code on behalf of allthingssecurity_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4294519302
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -40,15 +40,19 @@ public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger p
@Override
public void run() {
- if (processedEvents.get() > 0) {
- LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processedEvents.get());
+ // take the count when the checkpoint is started: an event that completes while it is being stored is counted
+ // for the next window from 0, and a checkpoint by batch size in the meantime cannot be confused with this one
+ final int processed = processedEvents.getAndSet(0);
+ if (processed > 0) {
+ LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processed);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
- error -> LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error),
- () -> {
- LOG.debug("Checkpoint updated.");
- processedEvents.set(0);
- });
+ error -> {
+ LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error);
+ // these events are not checkpointed: count them again for the next checkpoint
+ processedEvents.addAndGet(processed);
Review Comment:
Good catch, thanks. Changed the check to `cnt >= getConfiguration().getCheckpointBatchSize()` in 9fa055837ef6, with a comment on why. New test `batchSizeCheckpointStillWorksAfterFailedTimeoutCheckpoint` is your example: batch size 3, the timeout checkpoint of 2 events fails while one more event is counted, so the count is 3 and the next event takes it to 4. Without the change that event is not checkpointed by size (fails with "Wanted but not invoked"); with it, it is checkpointed at once with `COMPLETED_BY_SIZE`. camel-azure-eventhubs tests: 35 run, 0 failures.
_Claude Code on behalf of allthingssecurity_
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27922:
URL: https://github.com/apache/camel/pull/27922#discussion_r4293588144
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -40,15 +40,19 @@ public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger p
@Override
public void run() {
- if (processedEvents.get() > 0) {
- LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processedEvents.get());
+ // take the count when the checkpoint is started: an event that completes while it is being stored is counted
+ // for the next window from 0, and a checkpoint by batch size in the meantime cannot be confused with this one
+ final int processed = processedEvents.getAndSet(0);
+ if (processed > 0) {
+ LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processed);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
- error -> LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error),
- () -> {
- LOG.debug("Checkpoint updated.");
- processedEvents.set(0);
- });
+ error -> {
+ LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error);
+ // these events are not checkpointed: count them again for the next checkpoint
+ processedEvents.addAndGet(processed);
Review Comment:
Small edge case with the add-back: the size check in `EventHubsConsumer.processCommit` is `cnt == getConfiguration().getCheckpointBatchSize()`. Before this change the counter only went up by one, so it always hit the batch size exactly. Now a failed store can make it jump past it. For example, with batch size 3, `processed=2` and one event counted while the store was running: the counter is 1, the add-back makes it 3, and the next event makes it 4. From then on the `==` never matches, so the partition is checkpointed only by the timeout until the next timer run resets the counter. Using `cnt >= getConfiguration().getCheckpointBatchSize()` there would keep the size checkpoint working after a failed store.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org