Kafka消费者团体标识共享问题

Question

我有2个Kafka消费者共享相同的消费者组ID，但订阅不同的主题。它们中的每一个都只能从相应的主题中阅读。

当第一个使用者运行时，会从其订阅的主题中为其分配分区。当第二个使用者也运行时，使用者组会重新平衡（导致分配给第一个使用者的分区被撤销）。到目前为止，一切顺利。这与Kafka消费群体Id和消费者再平衡问题中的讨论一致。

但是，我开始在消费者1中看到TOPIC_AUTHORIZATION_FAILED——显然它试图访问它不订阅也无法访问的另一个主题。从这一点开始，消费者不会向前走，而是继续跑路。

我期望在重新平衡后，consumer 1将从它订阅的主题中重新分配分区，并继续前进。消费者1为什么尝试访问其他主题/我如何解决此问题？

日志

公共请求测试：消费者1的主题

公共请求开发：消费者2的主题

下面是Consumer 1日志（注意最后几行，它试图访问common request dev）-

{"@timestamp":"2021-10-22T07:39:17.550Z","message":"onPartitionsAssigned: [common-request-test-2, common-request-test-3, common-request-test-0, common-request-test-1]","severity":"INFO","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"c.m.l.e.s.a.config.KafkaReceiverConfig","ex":""}

{"@timestamp":"2021-10-22T07:39:17.853Z","message":"[Consumer clientId=consumer-stream-1, groupId=stream] Found no committed offset for partition common-request-test-2","severity":"INFO","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"o.a.k.c.c.internals.ConsumerCoordinator","ex":""}
{"@timestamp":"2021-10-22T07:39:17.857Z","message":"[Consumer clientId=consumer-stream-1, groupId=stream] Setting offset for partition common-request-test-3 to the committed offset FetchPosition{offset=5, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[kafka1-devtest1:9093 (id: 0 rack: null)], epoch=0}}","severity":"INFO","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"o.a.k.c.c.internals.ConsumerCoordinator","ex":""}
{"@timestamp":"2021-10-22T07:39:17.858Z","message":"[Consumer clientId=consumer-stream-1, groupId=stream] Setting offset for partition common-request-test-0 to the committed offset FetchPosition{offset=4, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[kafka1-devtest1:9093 (id: 0 rack: null)], epoch=0}}","severity":"INFO","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"o.a.k.c.c.internals.ConsumerCoordinator","ex":""}
{"@timestamp":"2021-10-22T07:39:17.858Z","message":"[Consumer clientId=consumer-stream-1, groupId=stream] Setting offset for partition common-request-test-1 to the committed offset FetchPosition{offset=4, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[kafka2-devtest1:9093 (id: 1 rack: null)], epoch=0}}","severity":"INFO","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"o.a.k.c.c.internals.ConsumerCoordinator","ex":""}
{"@timestamp":"2021-10-22T07:39:19.382Z","message":"[Consumer clientId=consumer-stream-1, groupId=stream] Resetting offset for partition common-request-test-2 to offset 0.","severity":"INFO","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"o.a.k.c.c.internals.SubscriptionState","ex":""}

{"@timestamp":"2021-10-22T07:43:21.598Z","message":"[Consumer clientId=consumer-stream-1, groupId=stream] Attempt to heartbeat failed since group is rebalancing","severity":"INFO","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"o.a.k.c.c.internals.AbstractCoordinator","ex":""}
{"@timestamp":"2021-10-22T07:43:21.599Z","message":"[Consumer clientId=consumer-stream-1, groupId=stream] Revoke previously assigned partitions common-request-test-2, common-request-test-3, common-request-test-0, common-request-test-1","severity":"INFO","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"o.a.k.c.c.internals.ConsumerCoordinator","ex":""}

{"@timestamp":"2021-10-22T07:43:21.599Z","message":"onPartitionsRevoked: [common-request-test-2, common-request-test-3, common-request-test-0, common-request-test-1]","severity":"INFO","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"c.m.l.e.s.a.config.KafkaReceiverConfig","ex":""}

{"@timestamp":"2021-10-22T07:43:21.599Z","message":"[Consumer clientId=consumer-stream-1, groupId=stream] (Re-)joining group","severity":"INFO","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"o.a.k.c.c.internals.AbstractCoordinator","ex":""}
{"@timestamp":"2021-10-22T07:43:22.174Z","message":"[Consumer clientId=consumer-stream-1, groupId=stream] Error while fetching metadata with correlation id 920 : {common-request-dev=TOPIC_AUTHORIZATION_FAILED}","severity":"WARN","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"org.apache.kafka.clients.NetworkClient","ex":""}

{"@timestamp":"2021-10-22T07:43:22.174Z","message":"[Consumer clientId=consumer-stream-1, groupId=stream] Topic authorization failed for topics [common-request-dev]","severity":"ERROR","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"org.apache.kafka.clients.Metadata","ex":""}

{"@timestamp":"2021-10-22T07:43:22.175Z","message":"[Consumer clientId=consumer-stream-1, groupId=stream] Join group failed with org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [common-request-dev]","severity":"INFO","service":"stream","x-correlation-id":"","client-id":"","account-id":"","trace":"0781f628a70f1e76","thread":"reactive-kafka-stream-1","class":"o.a.k.c.c.internals.AbstractCoordinator","ex":""}

Answer 1

听起来像您尝试使用StickyAssignor为您的消费者配置，或使用分配，而不是订阅（或与维护ACL策略的人交谈，并告诉他们您需要什么访问权限）

否则，是的，整个小组的主题集都会重新平衡。您正在为组订阅一个主题，而不是单个消费者。