Add parallel kafka consumer - #5
Conversation
📝 WalkthroughWalkthroughAdds a new Kafka parallel-consumer module with background polling, bounded record queues, acknowledgement-based offset commits, partition tracking, builder validation, and Testcontainers integration tests. It also updates log-generator APIs, dependencies, and Gradle project linkage. ChangesParallel Kafka Consumer
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Application
participant KafkaParallelConsumer
participant KafkaConsumer
participant Tracker
Application->>KafkaParallelConsumer: start()
KafkaParallelConsumer->>KafkaConsumer: subscribe(topic)
KafkaParallelConsumer->>KafkaConsumer: poll(pollTimeout)
KafkaConsumer-->>KafkaParallelConsumer: records
KafkaParallelConsumer->>Tracker: track(records)
KafkaParallelConsumer-->>Application: poll() returns record
Application->>KafkaParallelConsumer: ack(record)
KafkaParallelConsumer->>Tracker: complete(record)
Tracker->>KafkaConsumer: commitAsync(offset + 1)
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (2)
common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java (2)
15-16: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse Awaitility’s public package.
These imports couple the test to Testcontainers’ shaded implementation. Use
org.awaitility.Awaitilityfrom the declared Awaitility dependency instead; shaded packages are not stable APIs.Proposed import change
-import org.testcontainers.shaded.org.awaitility.Awaitility; -import org.testcontainers.shaded.org.awaitility.core.ThrowingRunnable; +import org.awaitility.Awaitility;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java` around lines 15 - 16, Update the imports in KafkaParallelConsumerTest to use Awaitility and ThrowingRunnable from the declared org.awaitility public package instead of org.testcontainers.shaded.org.awaitility, while leaving the test behavior unchanged.
47-59: 🗄️ Data Integrity & Integration | 🔵 Trivial | 🏗️ Heavy liftWait for asynchronous acknowledgements to be committed.
ack()only enqueues the record, but the test closes the consumer as soon as two records are observed. The tracking thread may not drain the acknowledgement queue, so this test can pass without validating offset commits. Wait for a commit-visible condition, such as restarting the same consumer group and asserting no redelivery, before closing.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java` around lines 47 - 59, Update the asynchronous test flow around consumer.ack(record) and consumer.close() so it waits for acknowledgements to be committed, not merely for both records to be observed. Before closing the consumer, verify commit visibility by restarting or reusing the same consumer group and asserting the acknowledged records are not redelivered, while preserving the existing log-content assertions.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java`:
- Around line 123-156: Update Builder.build() to default the Kafka consumer
property enable.auto.commit to false and reject configurations that explicitly
set it to true before constructing KafkaParallelConsumer. Preserve the existing
validation and ensure acknowledgement-driven offset handling cannot be used with
auto-commit enabled.
- Around line 78-84: The close() method must stop both workers and release the
Kafka client. After setting running false, wake and interrupt the poll and
tracking threads, join each non-null thread so trackingThread cannot remain
blocked in ackedRecordsQueue.take(), then call consumer.close() once the poll
loop has exited.
- Around line 106-111: Update handleUnackedMessages() to enqueue acknowledged
records for processing by the poll-owning thread instead of calling
tracker.track(record) directly. Ensure that thread invokes
Tracker.complete(record), including commitAsync(), while preserving the existing
running-loop and queue handling behavior.
In
`@common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java`:
- Line 29: Update KafkaParallelConsumerTest so sendRecord waits for each
asynchronous producer send to complete and propagates failures before returning.
Ensure test cleanup always closes both the consumer and kafkaProducer, including
failure paths, by moving cleanup into finally or an `@AfterEach` lifecycle method.
---
Nitpick comments:
In
`@common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java`:
- Around line 15-16: Update the imports in KafkaParallelConsumerTest to use
Awaitility and ThrowingRunnable from the declared org.awaitility public package
instead of org.testcontainers.shaded.org.awaitility, while leaving the test
behavior unchanged.
- Around line 47-59: Update the asynchronous test flow around
consumer.ack(record) and consumer.close() so it waits for acknowledgements to be
committed, not merely for both records to be observed. Before closing the
consumer, verify commit visibility by restarting or reusing the same consumer
group and asserting the acknowledged records are not redelivered, while
preserving the existing log-content assertions.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 3a5805a8-9c82-4058-ac18-8590dc4dfc5f
📒 Files selected for processing (10)
.idea/gradle.xmlcommon-libs/log-generator/build.gradlecommon-libs/log-generator/src/main/java/ir/pathlens/generator/CameraLogGenerator.javacommon-libs/parallel-consumer/build.gradlecommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.javacommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/PartitionTracker.javacommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/Tracker.javacommon-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.javacommon-libs/parallel-consumer/src/test/resources/application.ymlcommon-libs/test-extensions/build.gradle
| public void close() throws Exception { | ||
| running = false; | ||
| consumer.wakeup(); | ||
| if (pollThread != null) { | ||
| pollThread.interrupt(); | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
fd -a 'KafkaParallelConsumer\.java$' . || true
echo "== target outline =="
file="$(fd 'KafkaParallelConsumer\.java$' . | head -n1 || true)"
if [ -n "${file:-}" ]; then
echo "FILE=$file"
ast-grep outline "$file" --view expanded || true
echo "== target lines 1-140 =="
sed -n '1,140p' "$file" | nl -ba
fi
echo "== queue/thread references =="
rg -n "trackingThread|pollThread|ackedRecordsQueue|close\\(|consumer\\.close\\(|start\\(|join\\(\\)|interrupt\\(\\)" "$file" || trueRepository: alishazaee/PathLens
Length of output: 1243
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file="$(fd 'KafkaParallelConsumer\.java$' . | head -n1 || true)"
echo "FILE=$file"
echo "== lines 1-140 =="
sed -n '1,140p' "$file"
echo "== queue/thread/close references =="
rg -n "trackingThread|pollThread|ackedRecordsQueue|consumer\\.|close\\(|start\\(|join\\(\\)|interrupt\\(\\)" "$file" || true
echo "== Java version availability =="
command -v java || true
command -v javac || true
</file>Repository: alishazaee/PathLens
Length of output: 6685
🌐 Web query:
Apache Kafka KafkaConsumer close releases network connections and leaves consumer group
💡 Result:
Yes, calling close on a KafkaConsumer is required to properly release network connections and leave the consumer group [1][2][3]. When close is invoked, the consumer performs the following actions: 1. It attempts to commit any pending offsets (if auto-commit is enabled) [2]. 2. It sends a request to the group coordinator to gracefully leave the consumer group, which triggers an immediate rebalance [3]. This allows other members of the group to take over the partitions previously assigned to the closed consumer without waiting for the group coordinator to detect a timeout due to missing heartbeats [3]. 3. It closes the network connections (sockets) to the Kafka brokers to prevent resource leaks [1][3]. The close method has a default timeout (typically 30 seconds) to allow these cleanup tasks to complete [1][2][4]. If the consumer cannot gracefully leave the group or commit offsets within the specified timeout, it will be force-closed [2]. You should always invoke close after you are finished with a consumer instance to ensure resources are freed and the group rebalances promptly [1][3].
Citations:
- 1: https://kafka.apache.org/39/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html
- 2: https://kafka.apache.org/30/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html
- 3: https://stackoverflow.com/questions/51875216/kafkaconsumer-close-why
- 4: https://kafka.apache.org/38/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html
Finish both workers and close the Kafka client.
The current close() only wakes/interrupts the poll thread and leaves trackingThread stuck in ackedRecordsQueue.take() when no ack is queued. It also never calls consumer.close(), so broker connections and consumer-group membership are left to timeout. Join both worker threads and close the consumer after the poll loop exits.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java`
around lines 78 - 84, The close() method must stop both workers and release the
Kafka client. After setting running false, wake and interrupt the poll and
tracking threads, join each non-null thread so trackingThread cannot remain
blocked in ackedRecordsQueue.take(), then call consumer.close() once the poll
loop has exited.
| public Builder<K, V> withProperties(Properties properties) { | ||
| if (properties == null) { | ||
| throw new IllegalArgumentException("properties must not be null"); | ||
| } | ||
| this.properties = properties; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder<K, V> withQueueSize(int queueSize) { | ||
| if (queueSize <= 0) { | ||
| throw new IllegalArgumentException("queueSize must be positive, got: " + queueSize); | ||
| } | ||
| this.queueSize = queueSize; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder<K, V> withTopic(String topic) { | ||
| this.topic = topic; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder<K, V> withPollTimeout(Duration pollTimeout) { | ||
| if (pollTimeout == null || pollTimeout.isNegative()) { | ||
| throw new IllegalArgumentException("pollTimeout must be non-negative"); | ||
| } | ||
| this.pollTimeout = pollTimeout; | ||
| return this; | ||
| } | ||
|
|
||
| public KafkaParallelConsumer<K, V> build() { | ||
| Validate.notNull(properties, "properties must not be null"); | ||
| Validate.notNull(topic, "topic can not be null"); | ||
| Validate.isTrue(queueSize > 0, "queueSize must be positive"); | ||
| return new KafkaParallelConsumer<>(this); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate file =="
fd -a 'KafkaParallelConsumer.java' . || true
echo "== relevant outline =="
ast-grep outline common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java --view expanded || true
echo "== file section 1-260 =="
cat -n common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java | sed -n '1,260p'
echo "== search enable.auto.commit and build usages =="
rg -n "enable.auto.commit|auto.commit|withProperties|build\\(|ack\\(" common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer || trueRepository: alishazaee/PathLens
Length of output: 9614
🌐 Web query:
Apache Kafka 4.1 consumer config enable.auto.commit default true documentation
💡 Result:
The default value for the enable.auto.commit configuration in Apache Kafka 4.1 is true [1]. When set to true, the consumer's offset is periodically committed in the background [2][3], with the frequency of these commits controlled by the auto.commit.interval.ms configuration [2][3]. This behavior has remained consistent across Apache Kafka versions [1].
Citations:
- 1: https://kafka-options-explorer.conduktor.io/config/enable-auto-commit/
- 2: https://archive.apache.org/dist/kafka/4.1.1/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html
- 3: https://kafka.apache.org/43/configuration/consumer-configs/
Disable Kafka auto-commit for this acknowledgement-driven consumer.
enable.auto.commit defaults to true, so Kafka can commit offsets for polled records before the caller stores them in polledRecords and eventually calls ack(). Default it to false and reject an explicit true value during build().
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java`
around lines 123 - 156, Update Builder.build() to default the Kafka consumer
property enable.auto.commit to false and reject configurations that explicitly
set it to true before constructing KafkaParallelConsumer. Preserve the existing
validation and ensure acknowledgement-driven offset handling cannot be used with
auto-commit enabled.
a0db72d to
29d13d1
Compare
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java`:
- Around line 52-62: Update the rebalance callbacks so onPartitionsRevoked
commits the revoked partitions synchronously, then removes those same partitions
from tracker state via tracker.removePartitions(partitions). Do not remove
partitions in onPartitionsAssigned; preserve assignment tracking there so late
acknowledgements for revoked partitions are discarded.
- Around line 97-105: Update the poll loop in KafkaParallelConsumer so
drainCommittedMessages() executes once on every poll iteration, including when
ConsumerRecords is empty. Move or add the drain call outside the records loop
while preserving the existing per-record tracking and queue-offer behavior.
- Around line 49-67: Update KafkaParallelConsumer.start() to enforce a one-shot
lifecycle transition, preventing repeated calls from creating multiple poll
threads or polling the same consumer concurrently. Update close() to close the
consumer directly when start() was never completed, while preserving the
existing poll-thread shutdown and join behavior for started consumers.
In
`@common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/Tracker.java`:
- Around line 30-35: Tracker.complete must reject or discard acknowledgements
for offsets that were never registered or have already been completed, so
unknown offsets cannot advance commits. In
common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/Tracker.java
lines 30-35, retain delivered/in-flight offsets and complete each tracked offset
only once. In
common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/PartitionTrackerTest.java
lines 17-24, register offsets 102 and 103 for the out-of-order completion test
and add coverage proving an unregistered offset cannot advance the commit.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 8ffe3df2-1365-4dbb-8ca9-10591dbaab48
📒 Files selected for processing (11)
.idea/gradle.xmlcommon-libs/log-generator/build.gradlecommon-libs/log-generator/src/main/java/ir/pathlens/generator/CameraLogGenerator.javacommon-libs/parallel-consumer/build.gradlecommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.javacommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/PartitionTracker.javacommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/Tracker.javacommon-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.javacommon-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/PartitionTrackerTest.javacommon-libs/parallel-consumer/src/test/resources/application.ymlcommon-libs/test-extensions/build.gradle
🚧 Files skipped from review as they are similar to previous changes (7)
- common-libs/log-generator/build.gradle
- common-libs/parallel-consumer/src/test/resources/application.yml
- .idea/gradle.xml
- common-libs/parallel-consumer/build.gradle
- common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/PartitionTracker.java
- common-libs/test-extensions/build.gradle
- common-libs/log-generator/src/main/java/ir/pathlens/generator/CameraLogGenerator.java
29d13d1 to
cade1b2
Compare
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (1)
common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java (1)
24-24: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDeclare Awaitility as a direct test dependency.
KafkaParallelConsumerTestcallsAwaitility.await(), butparallel-consumeronly declares Testcontainers test dependencies, so the awaited import relies on a shaded implementation detail rather than a public API. Add an explicitorg.awaitilitytest dependency and importorg.awaitility.Awaitility.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java` at line 24, Update KafkaParallelConsumerTest to import org.awaitility.Awaitility instead of the Testcontainers-shaded Awaitility class, and add org.awaitility as a direct test dependency for parallel-consumer. Keep the existing Awaitility usage unchanged.Source: MCP tools
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In
`@common-libs/log-generator/src/main/java/ir/pathlens/generator/CameraLogGenerator.java`:
- Line 109: Restore backwards-compatible public access in CameraLogGenerator by
reintroducing generateLogProto(), randomIpv4(), randomIpv6(), and
randomPhoneNumber() as deprecated adapters that delegate to the current
implementations, preserving their prior behavior and signatures where possible.
Do not leave these API changes undocumented or inaccessible to external
consumers.
In
`@common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java`:
- Around line 99-106: Update the polling loop in KafkaParallelConsumer so
backpressure from polledRecords does not prevent consumer.poll() from running
beyond max.poll.interval.ms. Preserve delivery queue backpressure while
pausing/resuming assigned partitions, limiting each fetch to available capacity,
or using bounded enqueue waits that continue polling; ensure committed offsets
remain valid across rebalances.
- Around line 137-143: Update Builder.withProperties to preserve a
caller-provided ConsumerConfig.AUTO_OFFSET_RESET_CONFIG value by using
putIfAbsent instead of overwriting it, while retaining "earliest" only as the
default. Document that earliest is applied when no offset-reset policy is
supplied.
In
`@common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java`:
- Around line 76-90: Update the test around the generated records and Awaitility
assertion to retain the 50 original CameraLogProto.Log instances, then compare
them with the consumed logs rather than asserting only logs.size(). Use
multiset-style comparison if consumer ordering is not guaranteed, while
preserving the existing count and acknowledgment flow.
---
Nitpick comments:
In
`@common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java`:
- Line 24: Update KafkaParallelConsumerTest to import org.awaitility.Awaitility
instead of the Testcontainers-shaded Awaitility class, and add org.awaitility as
a direct test dependency for parallel-consumer. Keep the existing Awaitility
usage unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: a1ba1e43-1b53-4837-b4a6-7a44aa318061
📒 Files selected for processing (10)
.idea/gradle.xmlcommon-libs/log-generator/build.gradlecommon-libs/log-generator/src/main/java/ir/pathlens/generator/CameraLogGenerator.javacommon-libs/parallel-consumer/build.gradlecommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.javacommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/PartitionTracker.javacommon-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/Tracker.javacommon-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.javacommon-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/PartitionTrackerTest.javacommon-libs/test-extensions/build.gradle
🚧 Files skipped from review as they are similar to previous changes (3)
- common-libs/test-extensions/build.gradle
- .idea/gradle.xml
- common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/PartitionTrackerTest.java
| } | ||
|
|
||
| public CameraLogProto.Log.Builder generateLogProto() { | ||
| public CameraLogProto.Log.Builder generateLogBuilder() { |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -nP '\b(generateLogProto|randomIpv4|randomIpv6|randomPhoneNumber)\s*\(' --glob '*.java'Repository: alishazaee/PathLens
Length of output: 157
🏁 Script executed:
#!/bin/bash
set -e
# Locate and inspect the relevant file and signatures.
fd -a 'CameraLogGenerator\.java$' . | sed 's#^\./##'
file="$(fd 'CameraLogGenerator\.java$' . | head -n 1)"
if [ -n "$file" ]; then
wc -l "$file"
ast-grep outline "$file" --view expanded || true
echo '--- relevant sections ---'
sed -n '80,180p' "$file" | cat -n
fi
echo '--- search identifiers with surrounding context ---'
rg -nP '\bgenerateLogProto\b|\brandomIpv4\b|\brandomIpv6\b|\brandomPhoneNumber\b|\bgenerateLogBuilder\b|CameraLogGenerator\b' --glob '*.java' .Repository: alishazaee/PathLens
Length of output: 8508
Preserve backwards compatibility for the public CameraLogGenerator API.
CameraLogGenerator.generateLogProto() is no longer available, and randomIpv4(), randomIpv6(), and randomPhoneNumber() are now private static, so external consumers cannot call them. Add deprecated compatibility adapters or explicitly document/version this breaking change.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@common-libs/log-generator/src/main/java/ir/pathlens/generator/CameraLogGenerator.java`
at line 109, Restore backwards-compatible public access in CameraLogGenerator by
reintroducing generateLogProto(), randomIpv4(), randomIpv6(), and
randomPhoneNumber() as deprecated adapters that delegate to the current
implementations, preserving their prior behavior and signatures where possible.
Do not leave these API changes undocumented or inaccessible to external
consumers.
| ConsumerRecords<K, V> records = consumer.poll(pollTimeout); | ||
| for (ConsumerRecord<K, V> record : records) { | ||
| tracker.track(record); | ||
| drainCommittedMessages(); | ||
| while (!polledRecords.offer(record)) { | ||
| drainCommittedMessages(); | ||
| Thread.sleep(1); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C2 'max\.poll\.(interval|records)|while \(!polledRecords\.offer|consumer\.(pause|resume|poll)' \
common-libs/parallel-consumerRepository: alishazaee/PathLens
Length of output: 3255
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== file outline =="
ast-grep outline common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java || true
echo "== relevant source =="
cat -n common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java
echo "== config/package hints =="
rg -n "KafkaParallelConsumer|max.poll|max.poll.records|max.poll.interval|polledRecords|pause|resume|commit" common-libs/parallel-consumer/common-libs common-libs/parallel-consumer || trueRepository: alishazaee/PathLens
Length of output: 12634
🌐 Web query:
Apache Kafka KafkaConsumer poll max.poll.interval.ms consumer liveness documentation
💡 Result:
In Apache Kafka, max.poll.interval.ms is a configuration setting that defines the maximum allowed delay between consecutive invocations of the poll method when using consumer group management [1][2]. It serves as a liveness detection mechanism to prevent a consumer from holding onto assigned partitions indefinitely if it has stalled or is stuck in a livelock [3]. Key aspects of max.poll.interval.ms include: Mechanism: The KafkaConsumer proactively tracks the time elapsed between subsequent calls to poll. If this time exceeds the configured max.poll.interval.ms, the consumer considers itself failed, stops sending heartbeats, and voluntarily leaves the consumer group [3][4][5]. This triggers a rebalance so that the partitions can be reassigned to other healthy consumers [1][3]. Purpose and Liveness: This timeout is distinct from the session timeout (session.timeout.ms). While the session timeout monitors consumer health via background heartbeats to detect crashes or network issues, max.poll.interval.ms monitors the application's ability to process the records it has fetched [3][2][5]. It is specifically designed to handle scenarios where the consumer is technically alive (heartbeating in the background) but is not making progress in processing records [3]. Default Behavior and Tuning: - Default Value: The default is 300,000 milliseconds (5 minutes) [1][6][7]. - Triggering: Hitting this timeout typically indicates that your application code takes longer to process a batch of records than the allotted interval [4][5]. - Mitigation: To address frequent timeouts, you can increase max.poll.interval.ms to accommodate longer processing times or decrease max.poll.records to reduce the number of records returned in each poll, thereby shortening the time required to complete the poll loop [4][2][5]. Special Case for Static Membership: For consumers configured with a non-null group.instance.id (static members), reaching the max.poll.interval.ms timeout does not trigger an immediate partition reassignment [1]. Instead, the consumer stops sending heartbeats and the partitions will only be reassigned after the session timeout expires [1][4].
Citations:
- 1: https://kafka.apache.org/43/configuration/consumer-configs/
- 2: https://www.conduktor.io/kafka/kafka-consumer-important-settings-poll-and-internal-threads-behavior
- 3: https://kafka.apache.org/38/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html
- 4: https://cwiki.apache.org/confluence/display/KAFKA/KIP-517:+Add+consumer+metrics+to+observe+user+poll+behavior
- 5: https://stackoverflow.com/questions/72615626/kafka-consumer-death-handling
- 6: https://kafka.apache.org/41/configuration/consumer-configs/
- 7: https://kafka.apache.org/38/configuration/consumer-configs/
Keep poll() running while the delivery queue fills up.
The inner while (!polledRecords.offer(record)) loop can block for max.poll.interval.ms without calling consumer.poll() again, causing the consumer group to leave the assignment and making async commit offsets stale on rebalance. Pause/resume the assigned partitions while accepting backpressure, limit fetched batches to available queue capacity, or use bounded blocking enqueue behavior.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java`
around lines 99 - 106, Update the polling loop in KafkaParallelConsumer so
backpressure from polledRecords does not prevent consumer.poll() from running
beyond max.poll.interval.ms. Preserve delivery queue backpressure while
pausing/resuming assigned partitions, limiting each fetch to available capacity,
or using bounded enqueue waits that continue polling; ensure committed offsets
remain valid across rebalances.
| public Builder<K, V> withProperties(Properties properties) { | ||
| if (properties == null) { | ||
| throw new IllegalArgumentException("properties must not be null"); | ||
| } | ||
| this.properties = properties; | ||
| this.properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");; | ||
| return this; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C2 'AUTO_OFFSET_RESET_CONFIG|auto\.offset\.reset' common-libsRepository: alishazaee/PathLens
Length of output: 1776
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,190p' common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.javaRepository: alishazaee/PathLens
Length of output: 6370
Respect the caller’s offset-reset configuration.
withProperties() currently overwrites any explicit latest/none policy with earliest, so a new group can unexpectedly replay retained records. Use putIfAbsent(...) and document the default.
Proposed fix
- this.properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");;
+ this.properties.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| public Builder<K, V> withProperties(Properties properties) { | |
| if (properties == null) { | |
| throw new IllegalArgumentException("properties must not be null"); | |
| } | |
| this.properties = properties; | |
| this.properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");; | |
| return this; | |
| public Builder<K, V> withProperties(Properties properties) { | |
| if (properties == null) { | |
| throw new IllegalArgumentException("properties must not be null"); | |
| } | |
| this.properties = properties; | |
| this.properties.putIfAbsent(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); | |
| return this; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@common-libs/parallel-consumer/src/main/java/ir/pathlens/parallelconsumer/KafkaParallelConsumer.java`
around lines 137 - 143, Update Builder.withProperties to preserve a
caller-provided ConsumerConfig.AUTO_OFFSET_RESET_CONFIG value by using
putIfAbsent instead of overwriting it, while retaining "earliest" only as the
default. Document that earliest is applied when no offset-reset policy is
supplied.
| for (int i = 0; i < 50; i++) { | ||
| CameraLogProto.Log log = CameraLogGenerator.randomLog().generateLogBuilder().build(); | ||
| sendRecord(new ProducerRecord<>("test-2", log.toByteArray())); | ||
| } | ||
| consumer.start(); | ||
| List<CameraLogProto.Log> logs = new ArrayList<>(); | ||
| Awaitility.await() | ||
| .atMost(Duration.ofSeconds(10)) | ||
| .untilAsserted(() -> { | ||
| ConsumerRecord<byte[], byte[]> record = consumer.poll(); | ||
| if (record != null) { | ||
| logs.add(CameraLogProto.Log.parseFrom(record.value())); | ||
| consumer.ack(record); | ||
| } | ||
| assertEquals(50, logs.size()); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Assert the consumed records, not only their count.
Retain the 50 generated logs and compare them with the consumed logs (as a multiset if ordering is not guaranteed). The current assertion passes even when records are duplicated, stale, or otherwise incorrect.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@common-libs/parallel-consumer/src/test/java/ir/pathlens/parallelconsumer/KafkaParallelConsumerTest.java`
around lines 76 - 90, Update the test around the generated records and
Awaitility assertion to retain the 50 original CameraLogProto.Log instances,
then compare them with the consumed logs rather than asserting only logs.size().
Use multiset-style comparison if consumer ordering is not guaranteed, while
preserving the existing count and acknowledgment flow.
1 participant
Summary by CodeRabbit
New Features
Tests