CAMEL-24198: Track Event Hubs checkpoint batching per partition

[atiaomar1978-hub]

CAMEL-24198 is implemented, tested, committed, and pushed.

Issue

CAMEL-24198 — camel-azure-eventhubs - Redesign checkpoint store to use per-partition batching

The consumer used a single shared processedEvents, lastTask, and lastScheduledTask across all partitions, so checkpoint commits from different partitions could interfere with each other.

Fix

In EventHubsConsumer.java, replaced shared fields with per-partition maps:

• processedEventsByPartition
• checkpointTasksByPartition
• scheduledTasksByPartition

Each partition now tracks its own batch count and schedules its own timeout checkpoint independently.

Tests

Added EventHubsConsumerPerPartitionCheckpointTest.java with 4 tests:

• Per-partition event counters stay independent
• Batch-size checkpoint on one partition does not reset another
• Each partition gets its own scheduled checkpoint task
• Same partition reuses its checkpoint task within a batch window

12/12 tests passed (including existing shutdown and updater task tests).

Git

Item	Value
Branch	CAMEL-24198-eventhubs-per-partition-checkpoint-batching
Commit	8896d7755582
Remote	Pushed to origin

[gnodet]

Claude Code review on behalf of gnodet — AI-generated review

CAMEL-24198: Track Event Hubs checkpoint batching per partition

Verdict: APPROVE ✅

Fixes a real concurrency bug — the consumer used shared processedEvents, lastTask, and lastScheduledTask across all partitions, so checkpoint batching from different partitions interfered with each other (e.g., partition 0 reaching batch size could reset the counter that partition 1 was tracking).

What's good

• Correct data structure choice: ConcurrentHashMap for all three per-partition maps (processedEventsByPartition, checkpointTasksByPartition, scheduledTasksByPartition). Even though processCommit() is synchronized, the maps may be read from other paths, and ConcurrentHashMap is the right default for concurrent access.

• Minimal, focused change: Only the processCommit() method changes — the data structures shift from shared fields to per-partition maps, and the logic is straightforward: look up by partitionId, create if absent, operate on partition-local state.

• Null-check simplification is correct: lastTask != null && lastTask.isExpired() → checkpointTask.isExpired(). The null check was needed before because lastTask could be null from a prior invocation. Now checkpointTask is guaranteed non-null at the else if point — either just created (first branch) or fetched from the map (second branch).

• Improved debug logging: All log messages now include the partitionId, which is essential for diagnosing multi-partition issues in production.

• Good test coverage: 4 tests verify the key partition isolation properties — independent counters, batch-size checkpoint isolation, separate scheduled tasks, and task reuse within the same partition. The ArgumentCaptor verification that scheduled tasks are distinct instances is a nice touch.

Non-blocking observations

• Memory cleanup: The per-partition maps grow as partitions are seen but aren't explicitly cleared on doStop(). This is fine in practice since Event Hub partition counts are bounded (typically 2–32), but a clear() in doStop() would be tidy for completeness.

• Test assertions style: The tests use JUnit 5 assertEquals/assertNotSame rather than AssertJ. Not a blocker — both are valid in Camel, and the assertions are clear.

Clean fix for a real partition-interference bug. 👏

[atiaomar1978-hub]

Last change summary — commit 0905a931b031 on CAMEL-24198-eventhubs-per-partition-checkpoint-batching

What changed

Review feedback on the per-partition checkpoint batching work:

1. EventHubsConsumer.java

• In doStop(), clear all three per-partition maps:
  • processedEventsByPartition
  • scheduledTasksByPartition
  • checkpointTasksByPartition
• Avoids leaving partition state around after the consumer stops (partition count is small, but this keeps cleanup explicit).

2. EventHubsConsumerPerPartitionCheckpointTest.java

• Switched assertions from JUnit to AssertJ (assertThat(...).isEqualTo(), .isZero(), .hasSize(), .isEmpty(), .isNotSameAs()).
• Added doStopClearsPerPartitionCheckpointState to verify all maps are empty after doStop().

3. pom.xml

• Added assertj-core (test scope), aligned with other Azure components like camel-azure-servicebus.

Verification

• 20/20 EventHubs tests passed
• Pushed to origin/CAMEL-24198-eventhubs-per-partition-checkpoint-batching

Branch history

Commit	Description
8896d7755582	Initial per-partition checkpoint batching fix
0905a931b031	Review follow-up (cleanup + AssertJ + new test)

[gnodet]

Re-review of commit 2 (0905a93) — addresses review feedback

This commit cleanly addresses both non-blocking observations from my initial review:

1. Memory cleanup in doStop() ✅ — All three per-partition maps (processedEventsByPartition, scheduledTasksByPartition, checkpointTasksByPartition) are now cleared in doStop(), placed correctly after stopping the processorClient but before super.doStop(). This prevents stale state from accumulating across restart cycles.

2. AssertJ assertion style ✅ — All assertions migrated from JUnit assertEquals/assertNotSame to idiomatic AssertJ (assertThat(...).isEqualTo(...), .isZero(), .hasSize(), .isNotSameAs(), .isEmpty()). The assertj-core test dependency was added to pom.xml.

3. New test doStopClearsPerPartitionCheckpointState() ✅ — Good addition that verifies all three maps are empty after doStop() and confirms the processor client is stopped. The test properly sets up the processorClient mock via reflection and populates two partitions before stopping.

Approval from the previous review stands. Well done addressing the feedback!

---

Claude Code on behalf of gnodet — AI-generated review

[github-actions]

🌟 Thank you for your contribution to the Apache Camel project! 🌟
🤖 CI automation will test this PR automatically.

🐫 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 have access to the summary.

⚠️ Be careful when sharing logs. Review their contents before sharing them publicly.

[github-actions]

🧪 CI tested the following changed modules:

• components/camel-azure/camel-azure-eventhubs

---

🔬 Scalpel shadow comparison — Scalpel: 9 tested, 29 compile-only — current: 9 all tested

Maveniverse Scalpel detected 38 affected modules (current approach: 9).

⚠️ Modules only in Scalpel (29)

• apache-camel
• camel-allcomponents
• camel-catalog
• camel-catalog-console
• camel-catalog-lucene
• camel-catalog-maven
• camel-catalog-suggest
• camel-componentdsl
• camel-csimple-maven-plugin
• 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

Skip-tests mode would test 9 modules (1 direct + 8 downstream), skip tests for 29 (generated code, meta-modules)

Modules Scalpel would test (9)

• camel-azure-eventhubs
• camel-jbang-mcp
• camel-jbang-plugin-mcp
• camel-jbang-plugin-route-parser
• camel-jbang-plugin-tui
• camel-jbang-plugin-validate
• camel-launcher-container
• camel-yaml-dsl-validator
• camel-yaml-dsl-validator-maven-plugin

Modules with tests skipped (29)

• apache-camel
• camel-allcomponents
• camel-catalog
• camel-catalog-console
• camel-catalog-lucene
• camel-catalog-maven
• camel-catalog-suggest
• camel-componentdsl
• camel-csimple-maven-plugin
• 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

ℹ️ Shadow mode — Scalpel observes but does not affect test execution. Learn more

All tested modules (38 modules)

• Camel :: All Components Sync point
• Camel :: Assembly
• Camel :: Azure :: Event Hubs
• Camel :: Catalog :: CSimple Maven Plugin (deprecated)
• Camel :: Catalog :: Camel Catalog
• Camel :: Catalog :: Camel Report Maven Plugin
• Camel :: Catalog :: Camel Route Parser
• Camel :: Catalog :: Console
• Camel :: Catalog :: Dummy Component
• Camel :: Catalog :: Lucene (deprecated)
• Camel :: Catalog :: Maven
• Camel :: Catalog :: Suggest
• Camel :: Component DSL
• Camel :: Coverage
• Camel :: Docs
• Camel :: Endpoint DSL
• Camel :: Endpoint DSL :: Support
• Camel :: Integration Tests
• Camel :: JBang :: Core
• Camel :: JBang :: Integration tests
• Camel :: JBang :: MCP
• Camel :: JBang :: Main
• Camel :: JBang :: Plugin :: Edit
• Camel :: JBang :: Plugin :: Generate
• Camel :: JBang :: Plugin :: Kubernetes
• Camel :: JBang :: Plugin :: MCP
• Camel :: JBang :: Plugin :: Route Parser
• Camel :: JBang :: Plugin :: TUI
• Camel :: JBang :: Plugin :: Testing
• Camel :: JBang :: Plugin :: Validate
• Camel :: Kamelet Main
• Camel :: Launcher
• Camel :: Launcher :: Container
• Camel :: YAML DSL
• Camel :: YAML DSL :: Deserializers
• Camel :: YAML DSL :: Maven Plugins
• Camel :: YAML DSL :: Validator
• Camel :: YAML DSL :: Validator Maven Plugin

---

⚙️ View full build and test results