Add Kafka consumer record interception hook by graemerocher · Pull Request #1348 · micronaut-projects/micronaut-kafka

graemerocher · 2026-04-29T14:53:17Z

Summary

add a ConsumerRecordInterceptor API for Kafka listeners with listener-method matching and an InterceptionContext
apply matching interceptors before single-record and batch listener binding while preserving Micronaut-managed commit behavior for filtered records
document the new interception hook and add focused processor tests for ordering, filtering, context metadata, and method selection

Verification

./gradlew :micronaut-kafka:test --tests io.micronaut.configuration.kafka.processor.ConsumerStateSingleSpec --tests io.micronaut.configuration.kafka.processor.ConsumerStateBatchSpec --tests io.micronaut.configuration.kafka.processor.KafkaConsumerProcessorGracefulShutdownSpec
./gradlew publishGuide docs

Resolves #1204

Copilot

Copilot encountered an error and was unable to review this pull request. You can try again by re-requesting a review.

Copilot

Pull request overview

Copilot reviewed 11 out of 11 changed files in this pull request and generated 4 comments.

graemerocher · 2026-04-30T06:04:58Z

@copilot apply changes based on the comments in this thread

Copilot · 2026-04-30T06:12:33Z

@copilot apply changes based on the comments in this thread

Applied all four changes in 791d468:

Docs (kafkaListenerInterceptors.adoc): "a ann:..." → "an ann:..."
KafkaConsumerProcessor: TopicPartition in interceptRecords now built from intercepted.topic() / intercepted.partition() so wrapped records bucket correctly.
ConsumerStateBatch: Empty intercepted batch now calls handleResult(normalizeResult(null), consumerRecords) before returning, keeping the commit path consistent with successful processing.
ConsumerStateSingle: Hoisted interceptedConsumerRecord declaration before the try/catch; the catch block now selects the intercepted record (when non-null) for resolveWithErrorStrategy and resetTheFollowingPartitions so error metadata matches what was actually processed.

Warning

Firewall rules blocked me from connecting to one or more addresses (expand for details)

I tried to connect to the following addresses, but was blocked by firewall rules:

checkstyle.org
- Triggering command: /opt/hostedtoolcache/CodeQL/2.25.1/x64/codeql/tools/linux64/java/bin/java /opt/hostedtoolcache/CodeQL/2.25.1/x64/codeql/tools/linux64/java/bin/java -jar /opt/hostedtoolcache/CodeQL/2.25.1/x64/codeql/xml/tools/xml-extractor.jar --fileList=/tmp/codeql-scratch-5b054857fe035941/dbs/java/working/files-to-index16628508787889242430.list --sourceArchiveDir=/tmp/codeql-scratch-5b054857fe035941/dbs/java/src --outputDir=/tmp/codeql-scratch-5b054857fe035941/dbs/java/trap/java (dns block)
ge.micronaut.io
- Triggering command: /usr/lib/jvm/temurin-25-jdk-amd64/bin/java /usr/lib/jvm/temurin-25-jdk-amd64/bin/java --add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.lang.invoke=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.prefs/java.util.prefs=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.api=ALL-UNNAMED --add-exports=jdk.compiler/com.sun.tools.javac.util=ALL-UNNAMED --add-opens=java.base/java.util=ALL-UNNAMED --add-opens=java.prefs/java.util.prefs=ALL-UNNAMED --add-opens=java.base/java.nio.charset=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED --add-opens=java.base/java.util.concurrent=ALL-UNNAMED --add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED --add-opens=java.xml/javax.xml.namespace=ALL-UNNAMED --add-opens=java.base/java.time=ALL-UNNAMED --enable-native-access=ALL-UNNAMED -Xmx2048M -Dfile.encoding=UTF-8 -Duser.country -Duser.language=en (dns block)

If you need me to access, download, or install something from one of these locations, you can either:

Configure Actions setup steps to set up my environment, which run before the firewall is enabled
Add the appropriate URLs or hosts to the custom allowlist in this repository's Copilot coding agent settings (admins only)

Resolves #1204 Co-Authored-By: Codex with GPT-5 <codex@openai.com>

…it path, error record selection, and docs typo Agent-Logs-Url: https://github.com/micronaut-projects/micronaut-kafka/sessions/d513a696-a1bd-4396-a074-79da5f5ae36b Co-authored-by: graemerocher <66626+graemerocher@users.noreply.github.com>

Restore the multi-method ConsumerInfo constructor used by the new routing specs and stub pass-through interception in the state tests so they exercise the real dispatch path. Co-Authored-By: Codex with GPT-5 <codex@openai.com>

Co-Authored-By: Codex with GPT-5 <codex@openai.com>

sonarqubecloud · 2026-05-05T15:10:24Z

Quality Gate passed

Issues
6 New issues
0 Accepted issues

Measures
0 Security Hotspots
91.8% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

radovanradic · 2026-05-05T15:16:28Z

+        }
+        final ExecutableMethod<Object, ?> method = info.methodForTopic(topic);
+        Optional.ofNullable(info.ackArg(topic)).ifPresent(argument -> {
+            final Map<TopicPartition, OffsetAndMetadata> batchOffsets = getAckOffsets(interceptedTopicRecords);


Should it be getAckOffsets(originalTopicRecords)?

graemerocher added this to 6.0.0 Release Apr 29, 2026

graemerocher added the type: enhancement New feature or request label Apr 29, 2026

graemerocher self-assigned this Apr 29, 2026

graemerocher requested a review from Copilot April 29, 2026 14:59

Copilot AI reviewed Apr 29, 2026

Copilot started reviewing on behalf of graemerocher April 29, 2026 16:19 View session

graemerocher requested a review from Copilot April 29, 2026 16:34

graemerocher added this to 5.0.0 Release Apr 29, 2026

graemerocher removed this from 6.0.0 Release Apr 29, 2026

graemerocher moved this to Ready for review in 5.0.0 Release Apr 29, 2026

Copilot AI reviewed Apr 29, 2026

View reviewed changes

Copilot started reviewing on behalf of graemerocher April 29, 2026 17:10 View session

Copilot started work on behalf of graemerocher April 30, 2026 06:05 View session

Copilot finished work on behalf of graemerocher April 30, 2026 06:13

graemerocher force-pushed the issue-1204-wrap-record-consumption branch from 791d468 to f1759bf Compare April 30, 2026 07:55

graemerocher and others added 3 commits April 30, 2026 12:12

Add Kafka consumer record interception hook

a6d017b

Resolves #1204 Co-Authored-By: Codex with GPT-5 <codex@openai.com>

Fix consumer interception state test regressions

d50de0d

Restore the multi-method ConsumerInfo constructor used by the new routing specs and stub pass-through interception in the state tests so they exercise the real dispatch path. Co-Authored-By: Codex with GPT-5 <codex@openai.com>

graemerocher force-pushed the issue-1204-wrap-record-consumption branch from 9a97bcc to d50de0d Compare April 30, 2026 12:15

Fix Kafka scope consumer state test interception

0d77acf

Co-Authored-By: Codex with GPT-5 <codex@openai.com>

graemerocher requested a review from radovanradic April 30, 2026 13:01

radovanradic reviewed May 4, 2026

View reviewed changes

Comment thread kafka/src/main/java/io/micronaut/configuration/kafka/processor/ConsumerStateBatch.java Outdated

radovanradic reviewed May 4, 2026

View reviewed changes

Comment thread kafka/src/main/java/io/micronaut/configuration/kafka/processor/ConsumerStateBatch.java Outdated

radovanradic reviewed May 4, 2026

View reviewed changes

Comment thread kafka/src/main/java/io/micronaut/configuration/kafka/processor/ConsumerStateSingle.java Outdated

Address record interception PR feedback

e2d312d

Co-Authored-By: Codex with GPT-5 <codex@openai.com>

radovanradic reviewed May 5, 2026

View reviewed changes

sdelamo removed this from 5.0.0 Release May 7, 2026

sdelamo added this to 5.1.0 Release May 7, 2026

github-project-automation Bot moved this to Backlog in 5.1.0 Release May 7, 2026

graemerocher changed the base branch from 6.0.x to 6.1.x June 9, 2026 16:02

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Add Kafka consumer record interception hook#1348

Add Kafka consumer record interception hook#1348
graemerocher wants to merge 5 commits into
6.1.xfrom
issue-1204-wrap-record-consumption

graemerocher commented Apr 29, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

graemerocher commented Apr 30, 2026

Uh oh!

Copilot AI commented Apr 30, 2026 •

edited

Loading

I tried to connect to the following addresses, but was blocked by firewall rules:

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sonarqubecloud Bot commented May 5, 2026

Uh oh!

radovanradic May 5, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Uh oh!

Conversation

graemerocher commented Apr 29, 2026

Summary

Verification

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

graemerocher commented Apr 30, 2026

Uh oh!

Copilot AI commented Apr 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

I tried to connect to the following addresses, but was blocked by firewall rules:

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sonarqubecloud Bot commented May 5, 2026

Quality Gate passed

Uh oh!

radovanradic May 5, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Copilot AI commented Apr 30, 2026 •

edited

Loading