Skip to content

Fix pekko-connectors-kafka producer instrumentation on Scala 3 - #1432

Open
pjfanning wants to merge 2 commits into
kamon-io:masterfrom
pjfanning:fix-1416-pekko-kafka-advice
Open

pjfanning wants to merge 2 commits into
kamon-io:masterfrom
pjfanning:fix-1416-pekko-kafka-advice

Conversation

@pjfanning

@pjfanning pjfanning commented Oct 10, 2026 •

Copy link
Copy Markdown
Contributor

Fixes #1416.

Cause

ProduceWithEnvelopeContext was a plain object, and the advise(method, SomeObject) helper hands Kanela the generated class holding its static forwarders. On Scala 3 those forwarders keep the method annotations but drop the parameter annotations (confirmed with javap on 2.13 vs 3.3.5). With no annotation, ByteBuddy binds a parameter as @Advice.Argument(index), so exit(@Advice.Enter scope: Storage.Scope) was bound to produce's ProducerMessage$Envelope argument:

Cannot assign final org.apache.pekko.kafka.ProducerMessage$Envelope in to interface kamon.context.Storage$Scope

Changes

  • Make the advice a class + companion object with @static methods, advised via classOf[...], the same pattern the other Scala 3 instrumentations use. Parameter annotations are now present on both 2.13 and 3.
  • CommittingProducerSinkStageLogic was listed in the instrumentation but was missing from within in reference.conf, so it was never instrumented. I've added it. Its produce is private and compiles to org$apache$pekko$kafka$internal$CommittingProducerSinkStageLogic$$produce (same in connectors-kafka 1.0.0 and 1.2.0, Scala 2.13 and 3), so the method matcher now also accepts names ending in $$produce.

Testing

Added ProducerMessageInstrumentationSpec, which runs with the Kanela agent and needs no Kafka broker: the producer stages use Kafka's MockProducer (via ProducerSettings.withProducer), which records Kamon.currentContext() when send is called. It covers Message and MultiMessage through Producer.flexiFlow (DefaultProducerStageLogic) and a committable message through Producer.committableSink (CommittingProducerSinkStageLogic, with a fake committer from pekko-connectors-kafka-testkit).

Passes on Scala 2.13 and 3.3.5. Without the fix, it fails all three tests on Scala 3 (with the same Failed to transform error as in the issue) and the committable sink test on Scala 2.13.

🤖 Generated with Claude Code

pjfanning and others added 2 commits October 10, 2026 19:18
Scala 3 drops parameter annotations on the static forwarders of a plain
object, so ByteBuddy treated the exit advice's Storage.Scope parameter
as @argument(0) and failed to transform DefaultProducerStageLogic. Use a
class with @static methods, as the other Scala 3 instrumentations do.

Also instrument CommittingProducerSinkStageLogic, which was targeted but
missing from 'within' and whose private produce method is name-mangled.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Runs the producer stages against Kafka's MockProducer, so no broker is
needed, and checks the Context captured on message creation is current
when the stage calls send. Fails without the fix: all three tests on
Scala 3, and the committable sink test on Scala 2.13.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[kamon-pekko-connectors-kafka] Kamon 2.8.1 throws IllegalStateException when using pekko-connectors-kafka 1.1.0

1 participant