我试图使用< KStream> .process()与Time

Windows.of(“name”,30000)批量处理一些KTable值并发送它们.似乎30秒超过了消费者超时间隔,之后Kafka认为该消费者已经解散并释放分区.

我已经尝试提高轮询频率和提交间隔以避免这种情况：

config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "5000");

config.put(StreamsConfig.POLL_MS_CONFIG, "5000");

不幸的是,这些错误仍在发生：

(很多这些)

ERROR o.a.k.s.p.internals.RecordCollector - Error sending record to topic kafka_test1-write_aggregate2-changelog

org.apache.kafka.common.errors.TimeoutException: Batch containing 1 record(s) expired due to timeout while requesting metadata from brokers for kafka_test1-write_aggregate2-changelog-0

其次是：

INFO o.a.k.c.c.i.AbstractCoordinator - Marking the coordinator 12.34.56.7:9092 (id: 2147483547 rack: null) dead for group kafka_test1

WARN o.a.k.s.p.internals.StreamThread - Failed to commit StreamTask #0_0 in thread [StreamThread-1]:

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured session.timeout.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records.

at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:578)

显然,我需要更频繁地将心跳发送回服务器.怎么样？

我的拓扑结构是：

KStreamBuilder kStreamBuilder = new KStreamBuilder();

KStream lines = kStreamBuilder.stream(TOPIC);

KTable, String> kt = lines.aggregateByKey(

new DBAggregateInit(),

new DBAggregate(),

TimeWindows.of("write_aggregate2", 30000));

DBProcessorSupplier dbProcessorSupplier = new DBProcessorSupplier();

kt.toStream().process(dbProcessorSupplier);

KafkaStreams kafkaStreams = new KafkaStreams(kStreamBuilder, streamsConfig);

kafkaStreams.start();

KTable每隔30秒按键对值进行分组.在Processor.init()中,我调用context.schedule(30000).

DBProcessorSupplier提供DBProcessor的实例.这是AbstractProcessor的一个实现,其中提供了所有覆盖.他们只做LOG,所以我知道每个人都被击中.

这是一个非常简单的拓扑结构,但很明显我在某个地方错过了一个步骤.

编辑：

我知道我可以在服务器端进行调整,但我希望有一个客户端解决方案.我喜欢在客户端退出/死亡时很快就可以使用分区的概念.

编辑：

为了简化问题,我从图中删除了聚合步骤.它现在只是消费者 – >处理器(). (如果我将消费者直接发送到.print(),它会很快工作,所以我知道没关系). (类似地,如果我通过.print()输出聚合(KTable),它似乎也可以.

我发现.process() – 应该每隔30秒调用一次.punctuate()实际上阻塞了可变长度的时间并且随机输出(如果有的话).

进一步：

我将调试级别设置为’debug’并重新启动.我看到很多消息：

DEBUG o.a.k.s.p.internals.StreamTask - Start processing one record [ConsumerRecord

但是.punctuate()函数中的断点没有被击中.所以它做了很多工作,但没有让我有机会使用它.