登录
当前位置： 首页 > 知识库问答 >
问题：

Spring Kafka消费者与2.3.13.RELEASE和2.8.3版本之间的不兼容

卓瀚
2023-03-14

我在使用不同版本的 Spring-kafka 的应用程序的 kafka 使用者方面遇到了问题，特别是在 2.3.13.RELEASE 和 2.8.3 之间。

当使用蓝色/绿色策略部署到PRO时，我遇到的问题是，当在使用者上使用并发时，所有分区都被分配给使用版本2.3.13.RELEASE的应用程序，而在完成分区重新平衡时，使用Spring kafka版本2.8.3（同一主题和组名）部署使用者的新应用程序根本没有分区。

这是当前用于消费者的配置：

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

主题有10个分区。

以下是步骤中的应用程序日志：

    < li >使用spring-kafka版本2.3.13启动app1。向消费者发布:

APP1 日志

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: 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]

APP1 日志

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日志

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: 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-kafka 2.8.3 的版本）没有分配分区，而 APP1 保留在 10 个分配的分区中。

我已经尝试以相反的顺序部署，结果是一样的。

如果我部署具有相同版本的应用程序，则结果是正确的，每个应用程序分布 5 个分区。

我还尝试使用“RangeAsSignor”强制两个版本的分区策略，但我仍然有同样的问题。

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

[2022年5月17日编辑]

我用不同版本的SpringKafka做了更多的测试，最后，问题在SpringKafka2.3.13.RELEASE到SpringKafKa2.4.0.RELEASE的更新中重现。

根据2.4.0版本的发布公告：

2.4.0 kafka客户端与Spring for Apache Kafka 2.3不兼容，因此如果您希望使用2.4.0客户端，您必须升级到这个版本。有关如何覆盖jar版本的信息，请参见参考手册中的附录，尤其是当您使用Spring Boot进行依赖管理和/或使用测试嵌入式Kafka broker时。

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

共有1个答案

柳珂
2023-03-14

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

此外，kafka-客户端库已更新到这些Spring Kafka版本之间的新主要版本，因此这可能解释了行为上的差异-AFAIKSpring Kafka不会干扰分区分配，除非明确告知。

话虽如此，我不确定为什么您希望两个版本同时使用记录——如果业务逻辑发生变化，您将无法确定哪个版本将处理哪个分区/记录，除非您为每个版本手动分配分区。

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

您可以将此过程添加为蓝色/绿色部署自动化的一部分，如果有暴露的endpoint，可能会沿着负载均衡器路由更改添加。

有关管理容器生命周期的更多信息，请参阅SpringKafka文档的这一部分。

编辑：根据@garyrussell的评论，这个问题确实与Spring Kafka无关，而是与在kafka-client版本之间为消费者分配订阅ID的方式的变化有关，“范围分配器对它们进行排序，所有旧订阅都在列表中排在第一位”。

在此GitHub问题中全面讨论此事。

 类似资料：
  • Flink Kafka消费者和版本消费者之间的区别Flink Kafka消费者09/Flink Kafka消费者010/Flink Kafka消费者011

    Flink kafka消费者有两种类型的消费者，例如： 这两个消费者层次结构扩展了相同的类。我想知道维护编号类背后的设计决策是什么？我们什么时候应该使用其中一种？ 我注意到带有数字后缀的类有更多的特性（例如ratelimiting）。 https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka

  • SpringKafka消费者的异常

    在使用Spring Kafka Consumer时，我有时会收到以下错误消息。如代码片段所示，我至少实现了一次语义 1）我的疑问是，我是否错过了来自消费者的任何信息？ 2） 我需要处理这个错误吗。由于 org.apache.kafka.clients.consumer.提交失败异常：无法完成偏移提交，因为消费者不是自动分区分配的活动组的一部分；消费者很可能被踢出组。 我的SpringKafka消费

  • SpringKafka消费者再试

    我正在使用Spring Kafka consumer,它从主题中获取消息并将其保存到数据库中。如果满足故障条件，例如db不可用，kafka消费者库是否提供重试机制？如果是，是否有方法设置不同的重试间隔，如第1次重试应在5分钟后进行，第2次重试应在30分钟后进行，第3次重试应在1小时后进行等。

  • 2Kafka集群的SpringKafka消费者

    我在站点1（3个代理）有两个集群设置cluster-1，在站点2（3个代理）有两个集群设置cluster-2。使用spring kafka（1.3.6）消费者（一台机器）并通过@KafkaListener注释收听消息。我们如何为每个集群（c1和c2）实例化多个KafkaListenerContainerFactory，并同时监听来自这两个集群的数据。 我的侦听器应该同时使用来自这两个集群的消息。

  • 简述消费者与消费组之间的关系

    本文向大家介绍简述消费者与消费组之间的关系相关面试题，主要包含被问及简述消费者与消费组之间的关系时的应答技巧和注意事项，需要的朋友参考一下 消费者从属与消费组，消费偏移以消费组为单位。每个消费组可以独立消费主题的所有数据，同一消费组内消费者共同消费主题数据，每个分区只能被同一消费组内一个消费者消费。

  • SpringKafka消费者确认模式

    我需要使用consume process Product模式来处理Kafka消息，并已使用Kafka事务管理器配置了Spring Kafka侦听器容器，还设置了事务id前缀以启用Kafka事务。我正在使用批处理的ack模式，并试图了解在这种模式下，在事务中何时提交偏移量。文档似乎表明，一旦使用了轮询中的所有记录，ack模式批提交偏移量——在事务上下文中也是这样吗，即每个轮询1个事务？ 或者，在使用

相关问答

SpringKafka消费者确认模式Kafka消费者：消费者与消费者的抵消滞后Kafka10.2新消费者与旧消费者Kafka中消费者级别偏移和消费者组偏移之间的差异Spring security与Spring-4.3.0-release的兼容版本是什么？

相关文章

Java版本之间是否存在向后不兼容的特定示例？Eclipse Oxygen：不兼容的Java版本JMS-从一个消费者到多个消费者关于spring版本与JDK版本不兼容的问题及解决方法与Java 11兼容的最低Spring版本

相关阅读

消费者DefaultMQPushConsumer启动主要流程讯飞消费者java日常实习53min华为消费者bg 交互设计二面消费者负载均衡服务RebalanceService入口C#与Java之间的区别

相关工具

releaserelease-itPure Releasesemantic-releasenativescript-build-release

相关文档

Java API 版本的 Vert.x Core 手册略知 知识付费与内容变现 帮助文档Spring MVC 4.2.4.RELEASE 中文文档Effective Modern C++ 中文版（不完整）Sklearn 与 TensorFlow 机器学习实用指南第二版
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.