Spring Kafka Producer не отправляет в Kafka 1.0.0 (Magic v1 не поддерживает заголовки записей)

У меня была аналогичная проблема. Kafka добавляет заголовки по умолчанию, если мы используем JsonSerializer или JsonSerde для значений. Чтобы предотвратить эту проблему, нам нужно отключить добавление информационных заголовков.

если у вас все в порядке с сериализацией json по умолчанию, используйте следующее (ключевой момент здесь — ADD_TYPE_INFO_HEADERS):

Map<String, Object> props = new HashMap<>(defaultSettings);
props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
ProducerFactory<String, Object> producerFactory = new DefaultKafkaProducerFactory<>(props);

но если вам нужен пользовательский JsonSerializer с определенным ObjectMapper (например, с PropertyNamingStrategy.SNAKE_CASE), вы должны отключить явное добавление информационных заголовков в JsonSerializer, так как spring kafka игнорирует свойство DefaultKafkaProducerFactory ADD_TYPE_INFO_HEADERS (как по мне, это плохой дизайн spring kafka)

JsonSerializer<Object> valueSerializer = new JsonSerializer<>(customObjectMapper);
valueSerializer.setAddTypeInfo(false);
ProducerFactory<String, Object> producerFactory = new DefaultKafkaProducerFactory<>(props, Serdes.String().serializer(), valueSerializer);

или если мы используем JsonSerde, то:

Map<String, Object> jsonSerdeProperties = new HashMap<>();
jsonSerdeProperties.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
JsonSerde<T> jsonSerde = new JsonSerde<>(serdeClass);
jsonSerde.configure(jsonSerdeProperties, false);

Решено. Проблема не в брокере, не в каком-то кеше докеров и не в приложении Spring.

Проблема заключалась в потребителе консоли, который я использовал параллельно для отладки. Это был «старый» потребитель, начинавшийся с kafka-console-consumer.sh --topic=topic --zookeeper=...

На самом деле он выводит предупреждение при запуске: Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].

Следует использовать «нового» потребителя с параметром --bootstrap-server (особенно при использовании Kafka 1.0 с JsonSerializer). Примечание. Использование здесь старого потребителя действительно может повлиять на производителя.

Я только что провел тест на этом образе докера без проблем...

$docker ps

CONTAINER ID        IMAGE                    COMMAND                  CREATED             STATUS              PORTS                                                NAMES
f093b3f2475c        kafkadocker_kafka        "start-kafka.sh"         33 minutes ago      Up 2 minutes        0.0.0.0:32768->9092/tcp                              kafkadocker_kafka_1
319365849e48        wurstmeister/zookeeper   "/bin/sh -c '/usr/sb…"   33 minutes ago      Up 2 minutes        22/tcp, 2888/tcp, 3888/tcp, 0.0.0.0:2181->2181/tcp   kafkadocker_zookeeper_1

.

@SpringBootApplication
public class So47953901Application {

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

    @Bean
    public ApplicationRunner runner(KafkaTemplate<Object, Object> template) {
        return args -> template.send("foo", "bar", "baz");
    }

    @KafkaListener(id = "foo", topics = "foo")
    public void listen(String in) {
        System.out.println(in);
    }

}

.

spring.kafka.bootstrap-servers=192.168.177.135:32768
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false

.

2017-12-23 13:27:27.990  INFO 21305 --- [      foo-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [foo-0]
baz

ИЗМЕНИТЬ

У меня до сих пор работает...

spring.kafka.bootstrap-servers=192.168.177.135:32768
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

.

2017-12-23 15:27:59.997  INFO 44079 --- [           main] o.a.k.clients.producer.ProducerConfig    : ProducerConfig values: 
    acks = 1
    ...
    value.serializer = class org.springframework.kafka.support.serializer.JsonSerializer

...

2017-12-23 15:28:00.071  INFO 44079 --- [      foo-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [foo-0]
baz

вы используете версию kafka ‹=0.10.x.x, как только вы используете это, вы должны установить для JsonSerializer.ADD_TYPE_INFO_HEADERS значение false, как показано ниже.

Map<String, Object> props = new HashMap<>(defaultSettings);
props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
ProducerFactory<String, Object> producerFactory = new DefaultKafkaProducerFactory<>(props);

для вашего производителя заводских свойств.

Если вы используете версию kafka> 0.10.x.x, она должна работать нормально.