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

Reuse Header in AggregatingReplyingKafkaTemplate

Browse Source 
First of all we extract `Header correlation = record.headers().lastHeader(correlationHeaderName);`.

Then we convert its value back for a new header we add back to the aggregated records in the list.
Therefore no need in extra object and conversion logic.
This commit is contained in:
Nathan Xu
2023-12-13 10:35:57 -05:00
committed by GitHub
parent a98a7b8553
commit 7259e5eb5e
1 changed files with 1 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-kafka/src/main/java/org/springframework/kafka/requestreply/AggregatingReplyingKafkaTemplate.java
View File

@@ -34,7 +34,6 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.BatchConsumerAwareMessageListener;
@@ -148,11 +147,7 @@ public class AggregatingReplyingKafkaTemplate<K, V, R>
if (this.releaseStrategy.test(list, false)) {
ConsumerRecord<K, Collection<ConsumerRecord<K, R>>> done =
new ConsumerRecord<>(AGGREGATED_RESULTS_TOPIC, 0, 0L, null, list);
done.headers()
.add(new RecordHeader(correlationHeaderName,
isBinaryCorrelation()
? ((CorrelationKey) correlationId).getCorrelationId()
: ((String) correlationId).getBytes(StandardCharsets.UTF_8)));
done.headers().add(correlation);
this.pending.remove(correlationId);
checkOffsetsAndCommitIfNecessary(list, consumer);
completed.add(done);