[jira] [Commented] (KAFKA-14778) Kafka Streams 2.7.1 to 3.3.1 rolling upgrade with static membership triggers a rebalance

["Matthias J. Sax (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-14778?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17697181#comment-17697181 ] 

Matthias J. Sax commented on KAFKA-14778:
-----------------------------------------

Thanks for reporting this. – I believe the issue is as follows:
 * on restart, the consumer sends join group request (with laters subscription version)
 * broker returns buffered assignment without a rebalance (as it should)
 * KS inspects assignment and observes lower subscription – now things go "wrong"
 ** KS thinks that the assignment is invalid (because the leader should have sent an empty assignment on version probing only encoding it's version number)
 ** KS triggers a new rebalance sending it's subscription with lower subscription-version (as it assumes the leader could not decode the first subscription it sent)

It seem the fix might be to check if static membership is enabled and if the received assignment is empty of not. If static membership is enabled and if the assignment is not empty, it's not necessary to trigger a new rebalance.

One drawback is that all KS instances would stay on the old assignment, thus it might actually be desirable to have a single rebalance that allows all instances to switch to the new subscription/assignment version. This single rebalance should only happen _after_ all instances got updated. However, it's unclear how we could trigger such an rebalance, and it's also an open question if this rebalance is really desired or not?

> Kafka Streams 2.7.1 to 3.3.1 rolling upgrade with static membership triggers a rebalance
> ----------------------------------------------------------------------------------------
>
>                 Key: KAFKA-14778
>                 URL: https://issues.apache.org/jira/browse/KAFKA-14778
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>    Affects Versions: 2.7.1, 3.3.1
>            Reporter: Vinoth Rengarajan
>            Priority: Major
>
> Trying to upgrade Kaka Streams application from 2.7.1 to 3.3.1 with static membership but it triggers a rebalance
> Brokers are running on Kafka 2.7.1. Enabled the static membership in the application. Below are the configs {*}(Stream Config & Consumer Config){*}.
> Followed below steps to upgrade
>  * Brokers are running on Kafka 2.7.1(tried with 3.3.1 version then also rebalance happens).
>  * Application is running with 2.7.1 Kafka streams libraries.
>  * Deployed the latest version of the application with 3.3.1 Kafka streams libraries, and configured the *upgrade.from* property to 2.7 (based on the upgrade documentation available here [https://kafka.apache.org/33/documentation/streams/upgrade-guide]).
>  * Doing a rolling bounce with the latest changes, rebalance is being triggered on other instances in the cluster.
> Below are logs on the instance which is being bounced, forcing a rebalance on others. 
> *Logs:*
>  
> {code:java}
> INFO  2023-02-27 09:52:16.805 | streams.KafkaStreams stream-client [kafka_upgrade.Kafka_Upgrade_Test] State transition from CREATED to REBALANCING
> INFO  2023-02-27 09:52:16.946 | internals.ConsumerCoordinator [Consumer instanceId=kafka_upgrade.Kafka_Upgrade_Test-4, clientId=kafka_upgrade.Kafka_Upgrade_Test-StreamThread-4-consumer, groupId=kafka_upgrade.Kafka_Upgrade_Test] Notifying assignor about the new Assignment(partitions=[kafka_upgrade.Kafka_Upgrade_Test-version-updates-11, kafka_upgrade.Kafka_Upgrade_Test-version-updates-23], userDataSize=56)
> INFO  2023-02-27 09:52:16.947 | internals.StreamsPartitionAssignor stream-thread [kafka_upgrade.Kafka_Upgrade_Test-StreamThread-3-consumer] Sent a version 11 subscription and got version 8 assignment back (successful version probing). Downgrade subscription metadata to commonly supported version 8 and trigger new rebalance.
> INFO  2023-02-27 09:52:16.947 | internals.StreamsPartitionAssignor stream-thread [kafka_upgrade.Kafka_Upgrade_Test-StreamThread-2-consumer] Sent a version 11 subscription and got version 8 assignment back (successful version probing). Downgrade subscription metadata to commonly supported version 8 and trigger new rebalance.
> INFO  2023-02-27 09:52:16.947 | internals.StreamsPartitionAssignor stream-thread [kafka_upgrade.Kafka_Upgrade_Test-StreamThread-4-consumer] Sent a version 11 subscription and got version 8 assignment back (successful version probing). Downgrade subscription metadata to commonly supported version 8 and trigger new rebalance.
> INFO  2023-02-27 09:52:16.947 | internals.StreamsPartitionAssignor stream-thread [kafka_upgrade.Kafka_Upgrade_Test-StreamThread-1-consumer] Sent a version 11 subscription and got version 8 assignment back (successful version probing). Downgrade subscription metadata to commonly supported version 8 and trigger new rebalance.
> INFO  2023-02-27 09:52:16.947 | internals.StreamsPartitionAssignor stream-thread [kafka_upgrade.Kafka_Upgrade_Test-StreamThread-2-consumer] Requested to schedule immediate rebalance due to version probing.
> INFO  2023-02-27 09:52:16.948 | internals.StreamsPartitionAssignor stream-thread [kafka_upgrade.Kafka_Upgrade_Test-StreamThread-1-consumer] Requested to schedule immediate rebalance due to version probing.
> INFO  2023-02-27 09:52:16.948 | internals.StreamsPartitionAssignor stream-thread [kafka_upgrade.Kafka_Upgrade_Test-StreamThread-4-consumer] Requested to schedule immediate rebalance due to version probing.
> INFO  2023-02-27 09:52:16.948 | internals.StreamsPartitionAssignor stream-thread [kafka_upgrade.Kafka_Upgrade_Test-StreamThread-3-consumer] Requested to schedule immediate rebalance due to version probing. {code}
>  
>  
> *Streams Config:*
>  
> {code:java}
> acceptable.recovery.lag = 10000
> application.id = Kafka_Upgrade_Test
> application.server =
> bootstrap.servers = [broker1, broker2, broker3]
> buffered.records.per.partition = 1000
> built.in.metrics.version = latest
> cache.max.bytes.buffering = 10485760
> client.id = kafka_upgrade.Kafka_Upgrade_Test
> commit.interval.ms = 30000
> connections.max.idle.ms = 540000
> default.deserialization.exception.handler = class org.apache.kafka.streams.errors.LogAndFailExceptionHandler
> default.dsl.store = rocksDB
> default.key.serde = null
> default.list.key.serde.inner = null
> default.list.key.serde.type = null
> default.list.value.serde.inner = null
> default.list.value.serde.type = null
> default.production.exception.handler = class org.apache.kafka.streams.errors.DefaultProductionExceptionHandler
> default.timestamp.extractor = class org.apache.kafka.streams.processor.FailOnInvalidTimestamp
> default.value.serde = null
> max.task.idle.ms = 0
> max.warmup.replicas = 2
> metadata.max.age.ms = 300000
> metric.reporters = []
> metrics.num.samples = 2
> metrics.recording.level = INFO
> metrics.sample.window.ms = 30000
> num.standby.replicas = 0
> num.stream.threads = 4
> poll.ms = 100
> probing.rebalance.interval.ms = 600000
> processing.guarantee = at_least_once
> rack.aware.assignment.tags = []
> receive.buffer.bytes = 32768
> reconnect.backoff.max.ms = 1000
> reconnect.backoff.ms = 50
> repartition.purge.interval.ms = 30000
> replication.factor = 3
> request.timeout.ms = 120000
> retries = 0
> retry.backoff.ms = 100
> rocksdb.config.setter = null
> security.protocol = PLAINTEXT
> send.buffer.bytes = 131072
> state.cleanup.delay.ms = 600000
> state.dir = /mnt/store/yukon-apps
> task.timeout.ms = 300000
> topology.optimization = none
> upgrade.from = 2.7
> window.size.ms = null
> windowed.inner.class.serde = null
> windowstore.changelog.additional.retention.ms = 86400000 {code}
>  
>  
> *Consumer Config:*
>  
> {code:java}
> allow.auto.create.topics = true
> auto.commit.interval.ms = 10000
> auto.offset.reset = none
> bootstrap.servers = [server1, server2, server3]
> check.crcs = true
> client.dns.lookup = use_all_dns_ips
> client.id = kafka_upgrade.Kafka_Upgrade_Test-StreamThread-1
> client.rack =
> connections.max.idle.ms = 540000
> default.api.timeout.ms = 60000
> enable.auto.commit = false
> exclude.internal.topics = true
> fetch.max.bytes = 52428800
> fetch.max.wait.ms = 500
> fetch.min.bytes = 1
> group.id = Kafka_Upgrade_Test
> group.instance.id = Kafka_Upgrade_Test-1
> heartbeat.interval.ms = 5000
> interceptor.classes = []
> internal.leave.group.on.close = false
> internal.throw.on.fetch.stable.offset.unsupported = false
> isolation.level = read_uncommitted
> key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
> max.partition.fetch.bytes = 1048576
> max.poll.interval.ms = 1073741823
> max.poll.records = 2000
> metadata.max.age.ms = 300000
> metric.reporters = []
> metrics.num.samples = 2
> metrics.recording.level = INFO
> metrics.sample.window.ms = 30000
> partition.assignment.strategy = [org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor]
> receive.buffer.bytes = 65536
> reconnect.backoff.max.ms = 1000
> reconnect.backoff.ms = 50
> request.timeout.ms = 120000
> retry.backoff.ms = 100
> sasl.client.callback.handler.class = null
> sasl.jaas.config = null
> sasl.kerberos.kinit.cmd = /usr/bin/kinit
> sasl.kerberos.min.time.before.relogin = 60000
> sasl.kerberos.service.name = null
> sasl.kerberos.ticket.renew.jitter = 0.05
> sasl.kerberos.ticket.renew.window.factor = 0.8
> sasl.login.callback.handler.class = null
> sasl.login.class = null
> sasl.login.connect.timeout.ms = null
> sasl.login.read.timeout.ms = null
> sasl.login.refresh.buffer.seconds = 300
> sasl.login.refresh.min.period.seconds = 60
> sasl.login.refresh.window.factor = 0.8
> sasl.login.refresh.window.jitter = 0.05
> sasl.login.retry.backoff.max.ms = 10000
> sasl.login.retry.backoff.ms = 100
> sasl.mechanism = GSSAPI
> sasl.oauthbearer.clock.skew.seconds = 30
> sasl.oauthbearer.expected.audience = null
> sasl.oauthbearer.expected.issuer = null
> sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
> sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
> sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
> sasl.oauthbearer.jwks.endpoint.url = null
> sasl.oauthbearer.scope.claim.name = scope
> sasl.oauthbearer.sub.claim.name = sub
> sasl.oauthbearer.token.endpoint.url = null
> security.protocol = PLAINTEXT
> security.providers = null
> send.buffer.bytes = 131072
> session.timeout.ms = 300000
> socket.connection.setup.timeout.max.ms = 30000
> socket.connection.setup.timeout.ms = 10000
> ssl.cipher.suites = null
> ssl.enabled.protocols = [TLSv1.2]
> ssl.endpoint.identification.algorithm = https
> ssl.engine.factory.class = null
> ssl.key.password = null
> ssl.keymanager.algorithm = SunX509
> ssl.keystore.certificate.chain = null
> ssl.keystore.key = null
> ssl.keystore.location = null
> ssl.keystore.password = null
> ssl.keystore.type = JKS
> ssl.protocol = TLSv1.2
> ssl.provider = null
> ssl.secure.random.implementation = null
> ssl.trustmanager.algorithm = PKIX
> ssl.truststore.certificates = null
> ssl.truststore.location = null
> ssl.truststore.password = null
> ssl.truststore.type = JKS
> value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer {code}
>  
>  



--
This message was sent by Atlassian Jira
(v8.20.10#820010)