[
https://issues.apache.org/jira/browse/KAFKA-16514?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17838829#comment-17838829
]
Matthias J. Sax commented on KAFKA-16514:
-----------------------------------------
Thanks for the input. I was not reviewing/voting the original KIP nor the PR.
Thus, I did just assume there was some mentioning about static groups... As
there is nothing about it in the KIP as you pointed out, I did some digging and
the PR reveals why it's only implemented for static members:
[https://github.com/apache/kafka/pull/12035#discussion_r858263213]
We use admin client "removeMembersFromConsumerGroup" which only works for
static member, as it take the consumers `group.instance.id` as input. It seems
it was a pragmatic approach... Re-reading the KIP discussion it seems that
making it work for regular members would require a change in the consumer API,
and thus would have been a larger scope KIP (and the idea was to keep the KIP
scope limited).
Thus, while we might not need a KIP for Kafka Streams, we would need one for
the consumer to allow KS to use this newly added API... In the mean time, we
could still to a small PR to update the JavaDocs to call out the current
limitation (what will not make it a public contract IMHO, so after we get a
consumer KIP we can still address this limitation w/o another KIP).
Thoughts?
> Kafka Streams: stream.close(CloseOptions) does not respect options.leaveGroup
> flag.
> -----------------------------------------------------------------------------------
>
> Key: KAFKA-16514
> URL: https://issues.apache.org/jira/browse/KAFKA-16514
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 3.7.0
> Reporter: Sal Sorrentino
> Priority: Minor
>
> Working with Kafka Streams 3.7.0, but may affect earlier versions as well.
> When attempting to shutdown a streams application and leave the associated
> consumer group, the supplied `leaveGroup` option seems to have no effect.
> Sample code:
> {code:java}
> CloseOptions options = new CloseOptions().leaveGroup(true);
> stream.close(options);{code}
> The expected behavior here is that the group member would shutdown and leave
> the group, immediately triggering a consumer group rebalance. In practice,
> the rebalance happens after the appropriate timeout configuration has expired.
> I understand the default behavior in that there is an assumption that any
> associated StateStores would be persisted to disk and that in the case of a
> rolling restart/deployment, the rebalance delay may be preferable. However,
> in our application we are using in-memory state stores and standby replicas.
> There is no benefit in delaying the rebalance in this setup and we are in
> need of a way to force a member to leave the group when shutting down.
> The workaround we found is to set an undocumented internal StreamConfig to
> enforce this behavior:
> {code:java}
> props.put("internal.leave.group.on.close", true);
> {code}
> To state the obvious, this is less than ideal.
> Additional configuration details:
> {code:java}
> Properties props = new Properties();
> props.put(StreamsConfig.APPLICATION_ID_CONFIG, "someApplicationId");
> props.put(
> StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
> "localhost:9092,localhost:9093,localhost:9094");
> props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
> props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
> props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numProcessors);
> props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
> StreamsConfig.EXACTLY_ONCE_V2);{code}
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
