如何配置spring-kafka忽略错误格式的消息？

Question

我们的卡夫卡话题有一个问题，其中DefaultKafkaConsumerFactory & ConcurrentMessageListenerContainer组合描述here与工厂使用的JsonDeserializer消耗。不幸的是，有人有点热心，并发表了一些无效的消息到主题上。看来spring-kafka默默无法处理这些消息中的第一个。是否有可能让spring-kafka登录错误并继续？看看记录的错误消息，似乎Apache kafka-clients库应该处理这样的情况：当迭代一批消息时，它们中的一个或多个消息可能无法解析？

下面的代码是一个示例测试案例说明这个问题：

import org.apache.kafka.clients.consumer.ConsumerConfig; 
import org.apache.kafka.clients.consumer.ConsumerRecord; 
import org.apache.kafka.clients.producer.ProducerConfig; 
import org.apache.kafka.common.serialization.Serializer; 
import org.apache.kafka.common.serialization.StringDeserializer; 
import org.apache.kafka.common.serialization.StringSerializer; 
import org.junit.ClassRule; 
import org.junit.Test; 
import org.springframework.kafka.core.DefaultKafkaConsumerFactory; 
import org.springframework.kafka.core.DefaultKafkaProducerFactory; 
import org.springframework.kafka.core.KafkaTemplate; 
import org.springframework.kafka.listener.KafkaMessageListenerContainer; 
import org.springframework.kafka.listener.MessageListener; 
import org.springframework.kafka.listener.config.ContainerProperties; 
import org.springframework.kafka.support.SendResult; 
import org.springframework.kafka.support.serializer.JsonDeserializer; 
import org.springframework.kafka.support.serializer.JsonSerializer; 
import org.springframework.kafka.test.rule.KafkaEmbedded; 
import org.springframework.kafka.test.utils.ContainerTestUtils; 
import org.springframework.util.concurrent.ListenableFuture; 

import java.util.HashMap; 
import java.util.Map; 
import java.util.Objects; 
import java.util.concurrent.BlockingQueue; 
import java.util.concurrent.ExecutionException; 
import java.util.concurrent.LinkedBlockingQueue; 
import java.util.concurrent.TimeUnit; 
import java.util.concurrent.TimeoutException; 

import static org.junit.Assert.assertEquals; 
import static org.junit.Assert.assertThat; 
import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasKey; 
import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasValue; 

/** 
* @author jfreedman 
*/ 
public class TestSpringKafka { 
    private static final String TOPIC1 = "spring.kafka.1.t"; 

    @ClassRule 
    public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 1, TOPIC1); 

    @Test 
    public void submitMessageThenGarbageThenAnotherMessage() throws Exception { 
     final BlockingQueue<ConsumerRecord<String, JsonObject>> records = createListener(TOPIC1); 
     final KafkaTemplate<String, JsonObject> objectTemplate = createPublisher("json", new JsonSerializer<JsonObject>()); 

     sendAndVerifyMessage(records, objectTemplate, "foo", new JsonObject("foo"), 0L); 

     // push some garbage text to Kafka which cannot be marshalled, this should not interrupt processing 
     final KafkaTemplate<String, String> garbageTemplate = createPublisher("garbage", new StringSerializer()); 
     final SendResult<String, String> garbageResult = garbageTemplate.send(TOPIC1, "bar","bar").get(5, TimeUnit.SECONDS); 
     assertEquals(1L, garbageResult.getRecordMetadata().offset()); 

     sendAndVerifyMessage(records, objectTemplate, "baz", new JsonObject("baz"), 2L); 
    } 

    private <T> KafkaTemplate<String, T> createPublisher(final String label, final Serializer<T> serializer) { 
     final Map<String, Object> producerProps = new HashMap<>(); 
     producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString()); 
     producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, "TestPublisher-" + label); 
     producerProps.put(ProducerConfig.ACKS_CONFIG, "all"); 
     producerProps.put(ProducerConfig.RETRIES_CONFIG, 2); 
     producerProps.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1); 
     producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 5000); 
     producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5000); 
     producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); 
     producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, serializer.getClass()); 
     final DefaultKafkaProducerFactory<String, T> pf = new DefaultKafkaProducerFactory<>(producerProps); 
     pf.setValueSerializer(serializer); 
     return new KafkaTemplate<>(pf); 
    } 

    private BlockingQueue<ConsumerRecord<String, JsonObject>> createListener(final String topic) throws Exception { 
     final Map<String, Object> consumerProps = new HashMap<>(); 
     consumerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString()); 
     consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "TestConsumer"); 
     consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true); 
     consumerProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100"); 
     consumerProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15000); 
     consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); 
     consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class); 
     final DefaultKafkaConsumerFactory<String, JsonObject> cf = new DefaultKafkaConsumerFactory<>(consumerProps); 
     cf.setValueDeserializer(new JsonDeserializer<>(JsonObject.class)); 
     final KafkaMessageListenerContainer<String, JsonObject> container = new KafkaMessageListenerContainer<>(cf, new ContainerProperties(topic)); 
     final BlockingQueue<ConsumerRecord<String, JsonObject>> records = new LinkedBlockingQueue<>(); 
     container.setupMessageListener((MessageListener<String, JsonObject>) records::add); 
     container.setBeanName("TestListener"); 
     container.start(); 
     ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); 
     return records; 
    } 

    private void sendAndVerifyMessage(final BlockingQueue<ConsumerRecord<String, JsonObject>> records, 
             final KafkaTemplate<String, JsonObject> template, 
             final String key, final JsonObject value, 
             final long expectedOffset) throws InterruptedException, ExecutionException, TimeoutException { 
     final ListenableFuture<SendResult<String, JsonObject>> future = template.send(TOPIC1, key, value); 
     final ConsumerRecord<String, JsonObject> record = records.poll(5, TimeUnit.SECONDS); 
     assertThat(record, hasKey(key)); 
     assertThat(record, hasValue(value)); 
     assertEquals(expectedOffset, future.get(5, TimeUnit.SECONDS).getRecordMetadata().offset()); 
    } 

    public static final class JsonObject { 
     private String value; 

     public JsonObject() {} 

     JsonObject(final String value) { 
      this.value = value; 
     } 

     public String getValue() { 
      return value; 
     } 

     public void setValue(final String value) { 
      this.value = value; 
     } 

     @Override 
     public boolean equals(final Object o) { 
      if (this == o) { return true; } 
      if (o == null || getClass() != o.getClass()) { return false; } 
      final JsonObject that = (JsonObject) o; 
      return Objects.equals(value, that.value); 
     } 

     @Override 
     public int hashCode() { 
      return Objects.hash(value); 
     } 

     @Override 
     public String toString() { 
      return "JsonObject{" + 
        "value='" + value + '\'' + 
        '}'; 
     } 
    } 
} 

Answer 1

我有一个解决方案，但我不知道这是否是最好的之一，我扩展JsonDeserializer如下这会导致null值被spring-kafka消耗，并需要进行必要的下游更改以处理该情况。

class SafeJsonDeserializer[A >: Null](targetType: Class[A], objectMapper: ObjectMapper) extends JsonDeserializer[A](targetType, objectMapper) with Logging { 
    override def deserialize(topic: String, data: Array[Byte]): A = try { 
    super.deserialize(topic, data) 
    } catch { 
    case e: Exception => 
     logger.error("Failed to deserialize data [%s] from topic [%s]".format(new String(data), topic), e) 
     null 
    } 
}