将心跳间隔设置为小于会话超时的1/3。如果代理无法确定您的消费者是否还活着，它将在剩余消费者之间启动分区重新平衡。因此，您有一个心跳线程来通知代理，如果应用程序需要更长的时间来处理，那么使用者是活动的。在用户配置中更改这些：

heartbeat.interval.ms

session.timeout.ms

如果无法工作，请尝试增加会话超时。你必须摆弄这些价值观。

您可以在这里的活页夹级文档中设置任意kafka属性
spring.cloud.stream.kafka.binder.consumerproperties用户属性
任意kafka客户端消费者属性的键/值Map。除了支持已知的kafka消费属性，这里还允许未知的消费属性。这里的属性将取代在boot和上面的configuration属性中设置的任何属性。
默认值：空Map。
例如。 spring.cloud.stream.kafka.binder.consumerProperties.max.poll.records=10 或者在这里的绑定级文档。
spring.cloud.stream.kafka.bindings..consumer.configuration
使用包含通用kafka使用者属性的键/值对进行Map。除了具有kafka使用者属性外，还可以在此处传递其他配置属性。例如，应用程序需要的一些属性，如spring.cloud.stream.kafka.bindings.input.consumer.configuration.foo=bar。
默认值：空Map。
例如。 spring.cloud.stream.kafka.bindings.input.consumer.configuration.max.poll.records=10 您可以通过添加 OffsetCommitCallback 到侦听器容器的 ContainerProperties 和设置 syncCommits 至 false . 要自定义容器及其属性，请添加 ListenerContainerCustomizer 应用程序的bean。
编辑
异步提交回调。。。

@SpringBootApplication
@EnableBinding(Sink.class)
public class So57970152Application {

    public static void main(String[] args) {
        SpringApplication.run(So57970152Application.class, args);
    }

    @Bean
    public ListenerContainerCustomizer<AbstractMessageListenerContainer<byte[], byte[]>> customizer() {
        return (container, dest, group) -> {
            container.getContainerProperties().setAckMode(AckMode.RECORD);
            container.getContainerProperties().setSyncCommits(false);
            container.getContainerProperties().setCommitCallback((map, ex) -> {
                if (ex == null) {
                    System.out.println("Successful commit for " + map);
                }
                else {
                    System.out.println("Commit failed for " + map + ": " + ex.getMessage());
                }
            });
            container.getContainerProperties().setClientId("so57970152");
        };
    }

    @StreamListener(Sink.INPUT)
    public void listen(String in) {
        System.out.println(in);
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<byte[], byte[]> template) {
        return args -> {
            template.send("input", "foo".getBytes());
        };
    }

}

手动提交（同步）。。。

@SpringBootApplication
@EnableBinding(Sink.class)
public class So57970152Application {

    public static void main(String[] args) {
        SpringApplication.run(So57970152Application.class, args);
    }

    @Bean
    public ListenerContainerCustomizer<AbstractMessageListenerContainer<byte[], byte[]>> customizer() {
        return (container, dest, group) -> {
            container.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
            container.getContainerProperties().setClientId("so57970152");
        };
    }

    @StreamListener(Sink.INPUT)
    public void listen(String in, @Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment ack) {
        System.out.println(in);
        try {
            ack.acknowledge(); // MUST USE MANUAL_IMMEDIATE for this to work.
            System.out.println("Commit successful");
        }
        catch (Exception e) {
            System.out.println("Commit failed " + e.getMessage());
        }
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<byte[], byte[]> template) {
        return args -> {
            template.send("input", "foo".getBytes());
        };
    }

}