文档建议反馈控制台
首页
学习
活动
专区
圈层
工具
MCP广场
文章/答案/技术大牛
发布
社区首页 >问答首页 >Spring消费者与2.3.13版本之间的不兼容性

问Spring消费者与2.3.13版本之间的不兼容性
EN

Stack Overflow用户
提问于 2022-04-10 09:16:35
回答 1查看 561关注 0票数 0

我遇到了卡夫卡消费者的应用程序与不同版本的Spring，特别是2.3.13之间的problems和2.8.3。

当部署到PRO使用蓝色/绿色策略进行部署时，我遇到的问题是，当在使用者上使用并发性时，所有分区都被分配给使用2.3.13.RELEASE版本的应用程序，而在完成分区再平衡时部署带有Spring版本2.8.3 (相同主题和组名)的新应用程序则完全没有分区。

这是当前用于使用者的配置：

代码语言：javascript
复制
    @KafkaListener(topics = "spring.kafka.test", groupId = "group-spring-kafka", concurrency = "10")
    public void processMessage(String content) {
        System.out.println("Message received: " + content);
    }

主题有10个分区。

下面是分步骤的应用程序日志：

  1. 使用spring版本2.3.13启动app1：

APP1日志

代码语言：javascript
复制
  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::       (v2.2.13.RELEASE)

....9 thread consumer initialisations before this....

2022-04-10 10:56:44.516  INFO 2748 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 
    allow.auto.create.topics = true
    auto.commit.interval.ms = 5000
    auto.offset.reset = latest
    bootstrap.servers = [localhost:9092]
    check.crcs = true
    client.dns.lookup = default
    client.id = 
    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 = group-spring-kafka
    group.instance.id = null
    heartbeat.interval.ms = 3000
    interceptor.classes = []
    internal.leave.group.on.close = true
    isolation.level = read_uncommitted
    key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
    max.partition.fetch.bytes = 1048576
    max.poll.interval.ms = 300000
    max.poll.records = 500
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
    receive.buffer.bytes = 65536
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    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.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.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    send.buffer.bytes = 131072
    session.timeout.ms = 10000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = https
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLS
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
    value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

2022-04-10 10:56:44.519  INFO 2748 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 2.3.1
2022-04-10 10:56:44.519  INFO 2748 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: 18a913733fb71c01
2022-04-10 10:56:44.519  INFO 2748 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1649581004519
2022-04-10 10:56:44.520  INFO 2748 --- [           main] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-10, groupId=group-spring-kafka] Subscribed to topic(s): spring.kafka.test

.....

2022-04-10 10:56:44.520  INFO 2748 --- [           main] o.s.s.c.ThreadPoolTaskScheduler          : Initializing ExecutorService
2022-04-10 10:56:44.528  INFO 2748 --- [           main] com.spring.kafka.sample.DemoApplication  : Started DemoApplication in 1.858 seconds (JVM running for 2.218)
2022-04-10 10:56:47.670  INFO 2748 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-0]
2022-04-10 10:56:47.670  INFO 2748 --- [ntainer#0-3-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-4]
2022-04-10 10:56:47.670  INFO 2748 --- [ntainer#0-9-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-1]
2022-04-10 10:56:47.670  INFO 2748 --- [ntainer#0-2-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-3]
2022-04-10 10:56:47.670  INFO 2748 --- [ntainer#0-5-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-6]
2022-04-10 10:56:47.670  INFO 2748 --- [ntainer#0-1-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-2]
2022-04-10 10:56:47.670  INFO 2748 --- [ntainer#0-8-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-9]
2022-04-10 10:56:47.670  INFO 2748 --- [ntainer#0-6-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-7]
2022-04-10 10:56:47.670  INFO 2748 --- [ntainer#0-4-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-5]
  1. 使用Spring-KafkaVersion2.8.3启动app2 (与使用者app1相同的主题和组名)：

APP1日志

代码语言：javascript
复制
2022-04-10 11:00:02.696  INFO 2748 --- [ntainer#0-8-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-9, groupId=group-spring-kafka] Attempt to heartbeat failed since group is rebalancing
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-6-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-7, groupId=group-spring-kafka] Revoking previously assigned partitions [spring.kafka.test-7]
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-6-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions revoked: [spring.kafka.test-7]
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-8-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-9, groupId=group-spring-kafka] Revoking previously assigned partitions [spring.kafka.test-9]
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=group-spring-kafka] Attempt to heartbeat failed since group is rebalancing
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-6-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-7, groupId=group-spring-kafka] (Re-)joining group
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-1, groupId=group-spring-kafka] Revoking previously assigned partitions [spring.kafka.test-0]
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions revoked: [spring.kafka.test-0]
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-8-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions revoked: [spring.kafka.test-9]
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=group-spring-kafka] (Re-)joining group
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-8-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-9, groupId=group-spring-kafka] (Re-)joining group
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-9-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-10, groupId=group-spring-kafka] Attempt to heartbeat failed since group is rebalancing
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-9-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-10, groupId=group-spring-kafka] Revoking previously assigned partitions [spring.kafka.test-1]
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-9-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions revoked: [spring.kafka.test-1]
2022-04-10 11:00:02.697  INFO 2748 --- [ntainer#0-9-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-10, groupId=group-spring-kafka] (Re-)joining group
2022-04-10 11:00:02.699  INFO 2748 --- [ntainer#0-1-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=group-spring-kafka] Attempt to heartbeat failed since group is rebalancing
2022-04-10 11:00:02.699  INFO 2748 --- [ntainer#0-1-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-2, groupId=group-spring-kafka] Revoking previously assigned partitions [spring.kafka.test-2]
2022-04-10 11:00:02.699  INFO 2748 --- [ntainer#0-1-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions revoked: [spring.kafka.test-2]
2022-04-10 11:00:02.699  INFO 2748 --- [ntainer#0-1-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=group-spring-kafka] (Re-)joining group
2022-04-10 11:00:02.700  INFO 2748 --- [ntainer#0-3-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-4, groupId=group-spring-kafka] Attempt to heartbeat failed since group is rebalancing
2022-04-10 11:00:02.700  INFO 2748 --- [ntainer#0-3-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-4, groupId=group-spring-kafka] Revoking previously assigned partitions [spring.kafka.test-4]
2022-04-10 11:00:02.700  INFO 2748 --- [ntainer#0-3-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions revoked: [spring.kafka.test-4]
2022-04-10 11:00:02.700  INFO 2748 --- [ntainer#0-3-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-4, groupId=group-spring-kafka] (Re-)joining group
2022-04-10 11:00:02.701  INFO 2748 --- [ntainer#0-5-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-6, groupId=group-spring-kafka] Attempt to heartbeat failed since group is rebalancing
2022-04-10 11:00:02.701  INFO 2748 --- [ntainer#0-5-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-6, groupId=group-spring-kafka] Revoking previously assigned partitions [spring.kafka.test-6]
2022-04-10 11:00:02.701  INFO 2748 --- [ntainer#0-5-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions revoked: [spring.kafka.test-6]
2022-04-10 11:00:02.701  INFO 2748 --- [ntainer#0-5-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-6, groupId=group-spring-kafka] (Re-)joining group
2022-04-10 11:00:02.703  INFO 2748 --- [ntainer#0-4-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-5, groupId=group-spring-kafka] Attempt to heartbeat failed since group is rebalancing
2022-04-10 11:00:02.703  INFO 2748 --- [ntainer#0-4-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-5, groupId=group-spring-kafka] Revoking previously assigned partitions [spring.kafka.test-5]
2022-04-10 11:00:02.703  INFO 2748 --- [ntainer#0-7-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-8, groupId=group-spring-kafka] Attempt to heartbeat failed since group is rebalancing
2022-04-10 11:00:02.703  INFO 2748 --- [ntainer#0-4-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions revoked: [spring.kafka.test-5]
2022-04-10 11:00:02.703  INFO 2748 --- [ntainer#0-4-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-5, groupId=group-spring-kafka] (Re-)joining group
2022-04-10 11:00:02.703  INFO 2748 --- [ntainer#0-7-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-8, groupId=group-spring-kafka] Revoking previously assigned partitions [spring.kafka.test-8]
2022-04-10 11:00:02.703  INFO 2748 --- [ntainer#0-7-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions revoked: [spring.kafka.test-8]
2022-04-10 11:00:02.703  INFO 2748 --- [ntainer#0-7-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-8, groupId=group-spring-kafka] (Re-)joining group
2022-04-10 11:00:02.704  INFO 2748 --- [ntainer#0-2-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-3, groupId=group-spring-kafka] Attempt to heartbeat failed since group is rebalancing
2022-04-10 11:00:02.704  INFO 2748 --- [ntainer#0-2-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-3, groupId=group-spring-kafka] Revoking previously assigned partitions [spring.kafka.test-3]
2022-04-10 11:00:02.704  INFO 2748 --- [ntainer#0-2-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions revoked: [spring.kafka.test-3]
2022-04-10 11:00:02.704  INFO 2748 --- [ntainer#0-2-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-3, groupId=group-spring-kafka] (Re-)joining group
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-1, groupId=group-spring-kafka] Successfully joined group with generation 15
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-3-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-4, groupId=group-spring-kafka] Successfully joined group with generation 15
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-8-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-9, groupId=group-spring-kafka] Successfully joined group with generation 15
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-9-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-10, groupId=group-spring-kafka] Successfully joined group with generation 15
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-4-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-5, groupId=group-spring-kafka] Successfully joined group with generation 15
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-8-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-9, groupId=group-spring-kafka] Setting newly assigned partitions: spring.kafka.test-9
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-4-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-5, groupId=group-spring-kafka] Setting newly assigned partitions: spring.kafka.test-5
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-6-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-7, groupId=group-spring-kafka] Successfully joined group with generation 15
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-5-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-6, groupId=group-spring-kafka] Successfully joined group with generation 15
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-1-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=group-spring-kafka] Successfully joined group with generation 15
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-6-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-7, groupId=group-spring-kafka] Setting newly assigned partitions: spring.kafka.test-7
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-7-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-8, groupId=group-spring-kafka] Successfully joined group with generation 15
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-3-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-4, groupId=group-spring-kafka] Setting newly assigned partitions: spring.kafka.test-4
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-5-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-6, groupId=group-spring-kafka] Setting newly assigned partitions: spring.kafka.test-6
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-9-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-10, groupId=group-spring-kafka] Setting newly assigned partitions: spring.kafka.test-1
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-1, groupId=group-spring-kafka] Setting newly assigned partitions: spring.kafka.test-0
2022-04-10 11:00:02.708  INFO 2748 --- [ntainer#0-7-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-8, groupId=group-spring-kafka] Setting newly assigned partitions: spring.kafka.test-8
2022-04-10 11:00:02.707  INFO 2748 --- [ntainer#0-1-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-2, groupId=group-spring-kafka] Setting newly assigned partitions: spring.kafka.test-2
2022-04-10 11:00:02.710  INFO 2748 --- [ntainer#0-6-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-7, groupId=group-spring-kafka] Setting offset for partition spring.kafka.test-7 to the committed offset FetchPosition{offset=1, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=127.0.0.1:9092 (id: 1001 rack: null), epoch=0}}
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-1, groupId=group-spring-kafka] Setting offset for partition spring.kafka.test-0 to the committed offset FetchPosition{offset=1, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=127.0.0.1:9092 (id: 1001 rack: null), epoch=0}}
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-5-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-6, groupId=group-spring-kafka] Setting offset for partition spring.kafka.test-6 to the committed offset FetchPosition{offset=1, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=127.0.0.1:9092 (id: 1001 rack: null), epoch=0}}
2022-04-10 11:00:02.710  INFO 2748 --- [ntainer#0-2-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-3, groupId=group-spring-kafka] Successfully joined group with generation 15
2022-04-10 11:00:02.710  INFO 2748 --- [ntainer#0-3-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-4, groupId=group-spring-kafka] Setting offset for partition spring.kafka.test-4 to the committed offset FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=127.0.0.1:9092 (id: 1001 rack: null), epoch=0}}
2022-04-10 11:00:02.710  INFO 2748 --- [ntainer#0-7-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-8, groupId=group-spring-kafka] Setting offset for partition spring.kafka.test-8 to the committed offset FetchPosition{offset=1, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=127.0.0.1:9092 (id: 1001 rack: null), epoch=0}}
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-2-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-3, groupId=group-spring-kafka] Setting newly assigned partitions: spring.kafka.test-3
2022-04-10 11:00:02.710  INFO 2748 --- [ntainer#0-9-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-10, groupId=group-spring-kafka] Setting offset for partition spring.kafka.test-1 to the committed offset FetchPosition{offset=2, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=127.0.0.1:9092 (id: 1001 rack: null), epoch=0}}
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-6-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-7]
2022-04-10 11:00:02.710  INFO 2748 --- [ntainer#0-4-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-5, groupId=group-spring-kafka] Setting offset for partition spring.kafka.test-5 to the committed offset FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=127.0.0.1:9092 (id: 1001 rack: null), epoch=0}}
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-7-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-8]
2022-04-10 11:00:02.710  INFO 2748 --- [ntainer#0-8-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-9, groupId=group-spring-kafka] Setting offset for partition spring.kafka.test-9 to the committed offset FetchPosition{offset=1, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=127.0.0.1:9092 (id: 1001 rack: null), epoch=0}}
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-4-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-5]
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-5-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-6]
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-3-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-4]
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-8-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-9]
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-9-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-1]
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-1-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-2, groupId=group-spring-kafka] Setting offset for partition spring.kafka.test-2 to the committed offset FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=127.0.0.1:9092 (id: 1001 rack: null), epoch=0}}
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-1-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-2]
2022-04-10 11:00:02.711  INFO 2748 --- [ntainer#0-2-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-3, groupId=group-spring-kafka] Setting offset for partition spring.kafka.test-3 to the committed offset FetchPosition{offset=1, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=127.0.0.1:9092 (id: 1001 rack: null), epoch=0}}
2022-04-10 11:00:02.712  INFO 2748 --- [ntainer#0-2-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-3]
2022-04-10 11:00:02.750  INFO 2748 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: [spring.kafka.test-0]

APP2日志

代码语言：javascript
复制
  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::                (v2.6.4)

....9 thread consumer initialisations before this....

2022-04-10 11:00:02.096  INFO 2995 --- [           main] o.a.k.clients.consumer.ConsumerConfig    : ConsumerConfig values: 
    allow.auto.create.topics = true
    auto.commit.interval.ms = 5000
    auto.offset.reset = latest
    bootstrap.servers = [localhost:9092]
    check.crcs = true
    client.dns.lookup = use_all_dns_ips
    client.id = consumer-group-spring-kafka-10
    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 = group-spring-kafka
    group.instance.id = null
    heartbeat.interval.ms = 3000
    interceptor.classes = []
    internal.leave.group.on.close = true
    internal.throw.on.fetch.stable.offset.unsupported = false
    isolation.level = read_uncommitted
    key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
    max.partition.fetch.bytes = 1048576
    max.poll.interval.ms = 300000
    max.poll.records = 500
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor, class org.apache.kafka.clients.consumer.CooperativeStickyAssignor]
    receive.buffer.bytes = 65536
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    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.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.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    security.providers = null
    send.buffer.bytes = 131072
    session.timeout.ms = 45000
    socket.connection.setup.timeout.max.ms = 30000
    socket.connection.setup.timeout.ms = 10000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
    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.3
    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.StringDeserializer

assigned partitions: 
2022-04-10 11:00:02.719  INFO 2995 --- [ntainer#0-5-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: []
2022-04-10 11:00:02.719  INFO 2995 --- [ntainer#0-8-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: []
2022-04-10 11:00:02.719  INFO 2995 --- [ntainer#0-9-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: []
2022-04-10 11:00:02.719  INFO 2995 --- [ntainer#0-7-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: []
2022-04-10 11:00:02.719  INFO 2995 --- [ntainer#0-1-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: []
2022-04-10 11:00:02.719  INFO 2995 --- [ntainer#0-3-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: []
2022-04-10 11:00:02.719  INFO 2995 --- [ntainer#0-4-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: []
2022-04-10 11:00:02.719  INFO 2995 --- [ntainer#0-6-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: []
2022-04-10 11:00:02.719  INFO 2995 --- [ntainer#0-2-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: []
2022-04-10 11:00:02.719  INFO 2995 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : group-spring-kafka: partitions assigned: []

可以看到，APP2的日志(它是Spring-Kafka2.8.3的版本)没有分配分区，而APP1则保留在指定的10个分区中。

我试着按相反的顺序部署，结果是一样的。

如果我使用相同版本部署应用程序，结果是正确的，每个应用程序将分配5个分区。

我也尝试用"RangeAssignor“强制两个版本的分区策略，但是我仍然有同样的问题。

有什么想法吗？提前谢谢。

编辑5/17/2022

我已经用不同版本的Spring做了更多的测试，最终问题出现在从Spring版本2.3.13.RELEASE到Spring版本2.4.0.RELEASE的更新中。

根据发布的2.4.0发布公告

2.4.0Kafka客户端与2.3不兼容，因此如果您希望使用2.4.0客户端，则必须升级到此版本。有关如何覆盖jar版本，请参阅参考手册中的附录，特别是如果您使用Spring进行依赖关系管理和/或使用测试嵌入式Kafka broker。

我不确定..。广告中描述的不相容会导致这种行为吗？

spring-boot
concurrency
spring-kafka
consumer
rebalancing
EN

回答 1

Stack Overflow用户

发布于 2022-04-10 16:16:07

如果您希望两个应用程序之间的负载均衡，您应该将每个应用程序的并发性设置为5，否则，我不明白为什么会有任何保证，如果一个应用程序位于同一个用户组中，就不应该分配所有分区。

另外，kafka-clients库已经更新为这些Spring Kafka版本之间的一个新的主要版本，因此这可能解释了行为上的差异--除非明确告诉AFAIK Spring Kafka不干扰分区分配，否则AFAIK不会干扰分区分配。

尽管如此，我不知道为什么要同时使用两个版本的记录--如果业务逻辑有变化，您将无法确定地判断哪个版本将处理哪个分区/记录，除非您手动为每个版本分配分区。

您可以尝试的是--假设您正在部署green --在autoStartup设置为false的情况下部署它，确保应用程序部署正常，然后停止blue应用程序中的侦听器容器，并在green应用程序中启动容器。如果有任何问题发生，您可以停止green中的容器并重新启动blue容器作为回滚。

您可以将此过程添加为蓝色/绿色部署自动化的一部分，如果存在公开的端点，则可能沿负载均衡器路由更改。

有关管理容器生命周期的更多信息，请参阅Spring的本节文档。

编辑:根据@garyrussell的评论，这个问题实际上与Spring Kafka无关，而是因为在kafka-clients版本之间分配订阅id的方式发生了变化，“范围指定者对它们进行排序，所有旧的订阅都放在列表的第一位”。

在这个GitHub问题中对这个问题进行了充分的讨论。

票数 2
EN
页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：

https://stackoverflow.com/questions/71815245

复制
相关文章

相似问题

领券

腾讯云开发者

扫码关注腾讯云开发者

扫码关注腾讯云开发者

领取腾讯云代金券

热门产品

热门推荐

更多推荐

Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有 

深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 粤公网安备44030502008569号

腾讯云计算（北京）有限责任公司 京ICP证150476号 |  京ICP备11018762号

问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档