0

我在退订 Kafka 消费者时遇到问题，请注意我们使用的是 reactor kafka API。它在开始时确实成功取消订阅，但随后它立即加入组并分配了该主题的分区，因此基本上它一直保持订阅状态，并且即使不应该这样做，它也会继续消耗来自主题的消息！

以下是我做这项业务的代码，

import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.receiver.ReceiverRecord;

private final ReceiverOptions<String, byte[]> receiverOptions;
private Disposable disposable;

public void subscribe() {
    ReceiverOptions<String, byte[]> options = receiverOptions.subscription(Collections.singleton(topic))
            .addAssignListener(partitions -> {/*some listener code here*/})
            .addRevokeListener(partitions -> {/*some listener code here*/});

    Flux<ReceiverRecord<String, byte[]>> kafkaFlux = KafkaReceiver.create(options).receive();
    disposable = kafkaFlux
            .publishOn(Schedulers.fromExecutor(executorService))
            .subscribe(record -> {/*consume the record here*/});
}

public void unsubscribe() {
    if (disposable != null)
        disposable.dispose();
}

当我调用 unsubscribe() 方法时，如下日志所示；它撤销分区并向协调器发送离开组请求。

ConsumerCoordinator |[reactive-kafka-cgid-2]|| [Consumer clientId=consumer-cgid-2, groupId=cgid] Revoke previously assigned partitions topic-2
AbstractCoordinator |[reactive-kafka-cgid-2]|| [Consumer clientId=consumer-cgid-2, groupId=cgid] Member consumer-cgid-2-f61f347c-b07c-4037-92f5-9a418ac8d153 sending LeaveGroup request to coordinator kafaka_server:9084 (id: 2147483644 rack: null) due to the consumer is being closed

然而，在随后的一组事件发生并记录之后，该消费者被分配了主题分区并开始从那里消费消息。请注意，我没有在这里调用 subscribe 方法！

AbstractCoordinator |[reactive-kafka-cgid-1]|| [Consumer clientId=consumer-cgid-1, groupId=cgid] Attempt to heartbeat failed since group is rebalancing
ConsumerCoordinator |[reactive-kafka-cgid-1]|| [Consumer clientId=consumer-cgid-1, groupId=cgid] Revoke previously assigned partitions topic-0, topic-1
AbstractCoordinator |[reactive-kafka-cgid-1]|| [Consumer clientId=consumer-cgid-1, groupId=cgid] (Re-)joining group
AbstractCoordinator |[reactive-kafka-cgid-1]|| [Consumer clientId=consumer-cgid-1, groupId=cgid] Successfully joined group with generation Generation{generationId=3775, memberId='consumer-cgid-1-1f5da192-2a14-4634-a0f9-79707518598b', protocol='range'}
AbstractCoordinator |[reactive-kafka-cgid-1]|| [Consumer clientId=consumer-cgid-1, groupId=cgid] Successfully synced group in generation Generation{generationId=3775, memberId='consumer-cgid-1-1f5da192-2a14-4634-a0f9-79707518598b', protocol='range'}
ConsumerCoordinator |[reactive-kafka-cgid-1]|| [Consumer clientId=consumer-cgid-1, groupId=cgid] Notifying assignor about the new Assignment(partitions=[topic-0, topic-1, topic-2])
ConsumerCoordinator |[reactive-kafka-cgid-1]|| [Consumer clientId=consumer-cgid-1, groupId=cgid] Adding newly assigned partitions: topic-0, topic-2, topic-1
ConsumerCoordinator |[reactive-kafka-cgid-1]|| [Consumer clientId=consumer-cgid-1, groupId=cgid] Found no committed offset for partition topic-0
ConsumerCoordinator |[reactive-kafka-cgid-1]|| [Consumer clientId=consumer-cgid-1, groupId=cgid] Setting offset for partition topic-2 to the committed offset FetchPosition{offset=2049, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[kafaka_server:9084 (id: 3 rack: null)], epoch=14}}
ConsumerCoordinator |[reactive-kafka-cgid-1]|| [Consumer clientId=consumer-cgid-1, groupId=cgid] Setting offset for partition topic-1 to the committed offset FetchPosition{offset=119509, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[kafaka_server:9083 (id: 2 rack: null)], epoch=19}}
SubscriptionState   |[reactive-kafka-cgid-1]|| [Consumer clientId=consumer-cgid-1, groupId=cgid] Resetting offset for partition topic-0 to position FetchPosition{offset=2, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[kafaka_server:9082 (id: 1 rack: null)], epoch=22}}.

使用的版本是

org.apache.kafka:kafka-clients:2.8.0

org.apache.kafka:kafka-streams:2.8.0

io.projectreactor.kafka:reactor-kafka:1.1.0.RELEASE

值得注意的是，还有另一个服务正在为该服务正在使用的主题生成消息。

这可能与 kafka 参数有关，也可能与 kafka 参数无关，但如果您遇到这样的问题并且之前已经解决，请告诉我解决方案。

谢谢，

4

1 回答 1

0

我正在回答我自己的问题。

我编写了一个独立程序来订阅和取消订阅相关主题，并且该程序按预期工作。它清楚地表明从 Kafka 参数的角度来看根本没有问题（因此应用程序本身存在问题）。

在做了一些很好的代码分析并逐行查看代码之后，我注意到该subscribe方法被调用了 2 次。我commented是其中一个调用，然后进行了测试，它的表现都很好并且符合预期。

从来没有想过通过两次订阅主题，消费者将永远无法取消订阅！

注意 - 即使在执行 2 次调用之后unsubscribe，此使用者也不会取消订阅该主题。因此，如果它订阅了两次（或者更多可能 - 但我没有测试过），它将永远无法取消订阅！

从 Kafka 的角度来看，这是正常行为吗？我不太确定，保持这个项目开放让其他人回应......

谢谢...

于 2021-08-06T02:26:12.193 回答