You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mjsax (via GitHub)" <gi...@apache.org> on 2025/05/13 03:04:52 UTC
[PR] KAFKA-19271: Add internal ConsumerWrapper [kafka]
mjsax opened a new pull request, #19697:
URL: https://github.com/apache/kafka/pull/19697
With KIP-1071 enabled, the main consumer is created differently,
side stepping `KafkaClientSupplier`.
To allow injection test wrappers, we add an internal ConsumerWrapper,
until we define a new public interface.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19271: Add internal ConsumerWrapper [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #19697:
URL: https://github.com/apache/kafka/pull/19697#issuecomment-2876257325
You could consider enabling `KafkaStreamsTelemetryIntegrationTest` with the new wrapper for the new protocol to actually test that your changes work. Wdyt?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19271: Add internal ConsumerWrapper [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #19697:
URL: https://github.com/apache/kafka/pull/19697#discussion_r2086150911
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -550,12 +551,16 @@ private static MainConsumerSetup setupMainConsumer(final TopologyMetadata topolo
);
final ByteArrayDeserializer keyDeserializer = new ByteArrayDeserializer();
final ByteArrayDeserializer valueDeserializer = new ByteArrayDeserializer();
+
return new MainConsumerSetup(
- new AsyncKafkaConsumer<>(
- new ConsumerConfig(ConsumerConfig.appendDeserializerToConfig(consumerConfigs, keyDeserializer, valueDeserializer)),
- keyDeserializer,
- valueDeserializer,
- streamsRebalanceData
+ maybeWrapConsumer(
Review Comment:
I think this approach will break here:
https://github.com/apache/kafka/blame/trunk/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java#L1108
With the latest changes above, we need to make the extended subscribe method available to streams in the interface returned by the wrapper.
Could we introduce a little private interface `StreamsConsumer` that contains all extension methods for streams and let `AsyncKafkaConsumer` implement it, and have the wrapper return a `StreamsConsumer`?
This could serve as a basis for a public interface later on in KIP-1088.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org