I've got an integration tests that passes when I send to a Kafka topic without a key. However, when I add a key I start to get serialization errors.
org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition topic-1 at offset 0
Caused by: org.apache.kafka.common.errors.SerializationException: Size of data received by IntegerDeserializer is not 4
This is my sender class:
public class Sender {
private static final Logger LOG = LoggerFactory.getLogger(Sender.class);
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void send() {
String topic = "topic";
String data = "data";
String key = "key";
LOG.info("sending to topic: '{}', key: '{}', data: '{}'", topic, key, data);
// does not work
kafkaTemplate.send(topic, key, data);
// works
// kafkaTemplate.send(topic, data);
}
}
This is my configuration, where I specify a StringSerializer for the key
@Configuration
@EnableKafka
public class Config {
@Bean
public Sender sender() {
return new Sender();
}
@Bean
public Properties properties() {
return new Properties();
}
@Bean
public Map<String, Object> producerConfigs(Properties properties) {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServers());
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
@Bean
public ProducerFactory<String, String> producerFactory(Properties properties) {
return new DefaultKafkaProducerFactory<>(producerConfigs(properties));
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate(Properties properties) {
return new KafkaTemplate<>(producerFactory(properties));
}
}
The problem may related to the message listener in my test, but that is also using strings across the board
@RunWith(SpringRunner.class)
@SpringBootTest()
@DirtiesContext
public class SenderIT {
public static final Logger LOG = LoggerFactory.getLogger(SenderIT.class);
private static String SENDER_TOPIC = "topic";
@Autowired
private Sender sender;
private KafkaMessageListenerContainer<String, String> container;
private BlockingQueue<ConsumerRecord<String, String>> records;
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, SENDER_TOPIC);
@Before
public void setUp() throws Exception {
// set up the Kafka consumer properties
Map<String, Object> consumerProperties =
KafkaTestUtils.consumerProps("sender", "false", embeddedKafka);
consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
// create a Kafka consumer factory
DefaultKafkaConsumerFactory<String, String> consumerFactory =
new DefaultKafkaConsumerFactory<String, String>(consumerProperties);
// set the topic that needs to be consumed
ContainerProperties containerProperties = new ContainerProperties(SENDER_TOPIC);
// create a Kafka MessageListenerContainer
container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
// create a thread safe queue to store the received message
records = new LinkedBlockingQueue<>();
// setup a Kafka message listener
container.setupMessageListener(new MessageListener<String, String>() {
@Override
public void onMessage(ConsumerRecord<String, String> record) {
LOG.debug("test-listener received message='{}'", record.toString());
records.add(record);
}
});
// start the container and underlying message listener
container.start();
// wait until the container has the required number of assigned partitions
ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
}
@After
public void tearDown() {
// stop the container
container.stop();
}
@Test
public void test() throws InterruptedException {
sender.send();
// check that the message was received in Kafka
ConsumerRecord<String, String> kafkaTopicMsg = records.poll(10, TimeUnit.SECONDS);
LOG.debug("kafka recieved = {}", kafkaTopicMsg);
assertThat(kafkaTopicMsg).isNotNull();
}
}
As always, any help would be appreciated.
All the code to reproduce is available at https://github.com/LewisWatson/kafka-embedded-test/tree/8322621ad4e302d982e5ecd28af9fd314696d850
Full stack trace is available at https://travis-ci.org/LewisWatson/kafka-embedded-test/builds/273227986
After further inspection of the logs I was able to narrow the problem down to the test message listener
2017-09-08 09:30:06.845 ERROR 2550 --- [ -C-1] essageListenerContainer$ListenerConsumer : Container exception
org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition topic-1 at offset 0
Caused by: org.apache.kafka.common.errors.SerializationException: Size of data received by IntegerDeserializer is not 4
https://travis-ci.org/LewisWatson/kafka-embedded-test/builds/273227986#L2961
It looks like its expecting the key to be an integer for some reason.
Explicitly setting string deserializers for the consumer factory fixed the problem.
// create a Kafka consumer factory
DefaultKafkaConsumerFactory<String, String> consumerFactory =
new DefaultKafkaConsumerFactory<String, String>(consumerProperties,
new StringDeserializer(), new StringDeserializer());