如何使用spring集成kafka确认消费者读取的kafka消息

Question

我们使用 spring-integration-kafka 版本 3.1.2.RELEASE 和 int-kafka:message-driven-channel-adapter 来消费来自远程 kafka 主题的消息。生产者发送加密的消息，我们使用反序列化器解密实际的消息。我们能够使用该主题中发布的所有消息。我们将自动提交设置为 false。我们想知道在成功处理消息后如何提交或确认来自我们服务的消息。有人可以帮助我们如何提交从消息驱动通道读取的消息并提供一些参考实现吗？

当我们将自动提交设置为 true 时，我们假设它将在提交间隔后提交消息，但我们希望在我们的服务中处理它。我遇到过下面的示例，但我们在反序列化后收到一个自定义对象，而不是 Spring 集成消息。所以我们想知道如何在转换器中实现类似的确认，以便在转换过程中出现任何错误时我们不会提交消息。转换成功后提交消息。

  Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class); 
  if(acknowledgment != null) { System.out.println("Acknowledgment provided");
  acknowledgment.acknowledge(); }
   }



<int-kafka:message-driven-channel-adapter
    id="kafkaMessageListener"
    listener-container="kafkaMessageContainer" auto-startup="true"
    phase="100" send-timeout="5000" mode="record"
    message-converter="messageConverter"
    recovery-callback="recoveryCallback" error-message-strategy="ems"
    channel="inputFromKafkaChannel" error-channel="errorChannel" />

<int:transformer id="transformerid"
    ref="transformerBean"
    input-channel="inputFromKafkaChannel" method="transform"
    output-channel="messageTransformer" />

<bean id="container1" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
    <constructor-arg>
        <bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
            <constructor-arg>
                <map>
                    <entry key="bootstrap.servers" value="${spring.kafka.bootstrap-servers}" />
                    <entry key="enable.auto.commit" value="false" />
                    <entry key="auto.commit.interval.ms" value="100" />
                    <entry key="session.timeout.ms" value="15000" />
                    <entry key="group.id" value="${spring.kafka.consumer.group-id}" />
                    <entry key="key.deserializer"
                        value="org.apache.kafka.common.serialization.StringDeserializer" />
                    <entry key="value.deserializer"
                        value="com.test.CustomDeserializer" />
                </map>
            </constructor-arg>
        </bean>
    </constructor-arg>
    <constructor-arg>
        <bean class="org.springframework.kafka.listener.ContainerProperties">
            <constructor-arg name="topics" value="${spring.kafka.topics}" />
        </bean>
    </constructor-arg>
</bean>

Answer 1

auto.commit.offset=true意味着kafka-clients库提交了偏移量。

当为 false 时（对于 Apache Kafka 的 Spring 首选），侦听器容器在默认情况下接收到每个批次后提交偏移量poll()，但该机制由容器的AckMode属性控制。

请参阅提交偏移量。

如果您将容器设置AckMode为MANUAL或MANUAL_IMMEDIATE那么您的应用程序必须使用该对象执行提交Acknowledgment。

使用 Spring Integration 时，该Acknowledgment对象在标头中可用KafkaHeaders.ACKNOWLEDGMENT。

在大多数情况下，应该使用AckMode.BATCH（默认）或AckMode.RECORD，并且您的应用程序不需要担心提交偏移量。