Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-290: Add ProducerRecordCreator

Browse Source 
Resolves https://github.com/spring-projects/spring-integration-kafka/issues/290
This commit is contained in:
Gary Russell
2019-10-30 12:41:55 -04:00
committed by Artem Bilan
parent 6cddb2c169
commit cf3aa29a23
1 changed files with 45 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
View File

@@ -133,6 +133,10 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
private Type replyPayloadType = Object.class;
private ProducerRecordCreator<K, V> producerRecordCreator =
(message, topic, partition, timestamp, key, value, headers) ->
new ProducerRecord<>(topic, partition, timestamp, key, value, headers);
public KafkaProducerMessageHandler(final KafkaTemplate<K, V> kafkaTemplate) {
Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null");
this.kafkaTemplate = kafkaTemplate;
@@ -307,6 +311,16 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
this.replyPayloadType = payloadType;
}
/**
* Set a {@link ProducerRecordCreator} to create the {@link ProducerRecord}.
* @param producerRecordCreator the creator.
* @since 3.2.1
*/
public void setProducerRecordCreator(ProducerRecordCreator<K, V> producerRecordCreator) {
Assert.notNull(producerRecordCreator, "'producerRecordCreator' cannot be null");
this.producerRecordCreator = producerRecordCreator;
}
@Override
public String getComponentType() {
return this.isGateway ? "kafka:outbound-gateway" : "kafka:outbound-channel-adapter";
@@ -391,8 +405,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
headers = new RecordHeaders();
this.headerMapper.fromHeaders(messageHeaders, headers);
}
final ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partitionId, timestamp,
(K) messageKey, payload, headers);
final ProducerRecord<K, V> producerRecord = this.producerRecordCreator.create(message, topic, partitionId,
timestamp, (K) messageKey, payload, headers);
ListenableFuture<SendResult<K, V>> sendFuture;
RequestReplyFuture<K, V, Object> gatewayFuture = null;
if (this.isGateway) {
@@ -584,4 +598,33 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
}
/**
* Creates a {@link ProducerRecord} from a {@link Message} and/or properties
* derived from configuration and/or the message.
*
* @param <K> the key type.
* @param <V> the value type.
*
* @since 3.2.1
*
*/
@FunctionalInterface
public interface ProducerRecordCreator<K, V> {
/**
* Create a record.
* @param message the outbound message.
* @param topic the topic.
* @param partition the partition.
* @param timestamp the timestamp.
* @param key the key.
* @param value the value.
* @param headers the headers.
* @return the record.
*/
ProducerRecord<K, V> create(Message<?> message, String topic, Integer partition, Long timestamp, K key, V value,
Headers headers);
}
}