Construct different KafkaTemplate send same producerRecord, failure with java.lang.IllegalStateException: RecordHeaders has been closed.

[xzhan]

Problem:
we have two kafkaTemplates, one is for SJ to send ,second is for DFW to send message . then we use these two template to send same producer Record. the second Kafka template send message failure with java.lang.IllegalStateException: RecordHeaders has been closed.

Version:
Kafka-client-3.0.0
Spring-kafka 2.8.0

Stack:

java.lang.IllegalStateException: RecordHeaders has been closed.
at org.apache.kafka.common.header.internals.RecordHeaders.canWrite(RecordHeaders.java:122) ~[kafka-clients-3.0.0.jar:na]
at org.apache.kafka.common.header.internals.RecordHeaders.add(RecordHeaders.java:61) ~[kafka-clients-3.0.0.jar:na]
at org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.addHeader(AbstractJavaTypeMapper.java:145) ~[spring-kafka-2.8.0.jar:2.8.0]
at org.springframework.kafka.support.mapping.DefaultJackson2JavaTypeMapper.fromJavaType(DefaultJackson2JavaTypeMapper.java:172) ~[spring-kafka-2.8.0.jar:2.8.0]
at org.springframework.kafka.support.serializer.JsonSerializer.serialize(JsonSerializer.java:201) ~[spring-kafka-2.8.0.jar:2.8.0]
at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:929) ~[kafka-clients-3.0.0.jar:na]
at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:889) ~[kafka-clients-3.0.0.jar:na]
at org.springframework.kafka.core.DefaultKafkaProducerFactory$CloseSafeProducer.send(DefaultKafkaProducerFactory.java:984) ~[spring-kafka-2.8.0.jar:2.8.0]
at org.springframework.kafka.core.KafkaTemplate.doSend(KafkaTemplate.java:649) ~[spring-kafka-2.8.0.jar:2.8.0]

Debug
second time to send the producer Record with isReadOnly=true, I don't know when set it as true from false

@garyrussell can you help to check the issue? or my wrong usage ?

[artembilan]

This has nothing to do with Spring for Apache Kafka. Please, consider to ask questions in the future on StackOverflow and be sure that the problem is really relevant somehow to Spring for Apache Kafka.

What you are facing is a standard immutability of ProducerRecord when it has been sent already, "closed" in terms of Kafka Client.
You need to consider to create a new ProducerRecord for each template, or don't create it at all and let the template to create it on your behalf internally from the info you have provided.
See other template methods:

	/**
	 * Send the data to the provided topic with the provided key and partition.
	 * @param topic the topic.
	 * @param partition the partition.
	 * @param timestamp the timestamp of the record.
	 * @param key the key.
	 * @param data the data.
	 * @return a Future for the {@link SendResult}.
	 * @since 1.3
	 */
	ListenableFuture<SendResult<K, V>> send(String topic, Integer partition, Long timestamp, K key, V data);
...
	/**
	 * Send a message with routing information in message headers. The message payload
	 * may be converted before sending.
	 * @param message the message to send.
	 * @return a Future for the {@link SendResult}.
	 * @see org.springframework.kafka.support.KafkaHeaders#TOPIC
	 * @see org.springframework.kafka.support.KafkaHeaders#PARTITION_ID
	 * @see org.springframework.kafka.support.KafkaHeaders#MESSAGE_KEY
	 */
	ListenableFuture<SendResult<K, V>> send(Message<?> message);

With the Message you can achieve a goal to transfer some custom headers. The DefaultKafkaHeaderMapper does the trick for us to map headers from the message to the ProducerRecord.
See more in docs: https://docs.spring.io/spring-kafka/docs/current/reference/html/#sending-messages

GitHub issues are really for bug reports and new feature requests.
We just don't handle support over here.
Better to ask for a help in the community forums, so you may get an answer faster than we wake up in the morning next day.

[garyrussell]

ProducerRecord must be immutable due to the asynchronous nature of Kafka publishing.

[xzhan]

thx @artembilan and @garyrussell , Educated

[xzhan]

I have follow @artembilan suggestion , @artembilan as we generate the ProducerRecord directly from consumerRecord, that's why I construct it with one time as all routing DC share same topic name.