Kafka消费者-暂停特定Kafka主题分区的事件轮询，以将其用作延迟队列

Question

在我们的系统中有一个场景，kafka主题XYZ用户详细信息由某个其他生产应用程序A(不同的系统)发布，而我的应用程序B正在使用该主题。

要求是应用程序B需要消耗该事件45分钟后（或任何可配置的时间），它被放在kafka主题XYZ由A（这种延迟的原因是，一些系统C的另一个REST api需要根据特定用户的此用户详细信息事件触发，以确认它是否为该用户设置了一些标志，并且该标志可以在45分钟持续时间内的任何点设置，尽管如果C没有能力发布到kafka或以任何方式通知我们，它可能已经解决）。

我们的应用程序B是在Spring编写的。

我尝试的解决方案是从Kafka获取事件，并检查队列中第一个事件的时间戳，如果该事件已经45分钟，则处理该事件，如果该时间小于45分钟，那么使用MessageListnerContainer pause（）方法暂停轮询Kafka容器一段时间，直到达到45分钟。类似以下内容-

@KafkaListener(id = "delayed_listener", topics = "test_topic", groupId = "test_group")
        public void delayedConsumer(@Payload  String message,
                                    Acknowledgment acknowledgment) {

            UserDataEvent userDataEvent = null;
            try {
                 userDataEvent = this.mapper.readValue(message, TopicRequest.class);
            } catch (JsonProcessingException e) {
                logger.error("error while parsing message");
            }
            MessageListenerContainer delayedContainer = this.kafkaListenerEndpointRegistry.getListenerContainer("delayed_listener");
            if (userDataEvent.getPublishTime() > 45 minutes) // this will be some configured value
 {
                long sleepTimeForPolling = userDataEvent.getPublishTime() - System.currentTimeMillis();
                // give negative ack to put already polled messages back to kafka topic
                acknowledgment.nack(1000);
                // pause container, and later resume it  
                delayedContainer.pause();
                ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
                scheduledExecutorService.schedule(() -> {
                    delayedContainer.resume();
                }, sleepTimeForPolling, TimeUnit.MILLISECONDS);
                return;
            }
            // if message was already 45 minutes old then process it
            this.service.processMessage(userDataEvent);
            acknowledgment.acknowledge();
        }

虽然它适用于单个分区，但我不确定这是否是一个正确的方法，对此有什么评论吗？我还看到多个分区，它会导致问题，因为上面的暂停方法调用将暂停整个容器，如果其中一个分区有旧消息，如果容器因其他分区中的新消息而暂停，它将不会被消费。我可以在分区级别以某种方式使用此暂停逻辑吗？

在一定数量的可配置时间后，有什么更好/推荐的解决方案来实现这种延迟处理，在这种情况下我可以采用这些时间，而不是做我上面所做的？

Answer 1

Kafka并不是真的为这样的场景设计的。

我可以看到这种技术工作的一种方式是将容器并发设置为与主题中的分区数量相同，以便每个分区由不同线程上的不同消费者处理；然后暂停/恢复单个消费者

为此，请添加 消费者