commit c9f8aa960aae4aa157152a9f69a828f7ac7690dd
author Tzu-Li (Gordon) Tai <tzulitai@apache.org> Wed Apr 12 07:51:21 2023 -0700
committer Tzu-Li (Gordon) Tai <tzulitai@apache.org> Wed Apr 12 07:51:21 2023 -0700
tree dca587403796f8000d50bd4259a7941130e49de2
parent a7785630e714af303b224c38d9a6caa89a551265

[hotfix] [flink-connector-kafka] Simplify ProducerRecord instantiation w.r.t. headers

flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java

diff --git a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java
index 1cc9220..34cf6ef 100644
--- a/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java
+++ b/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaRecordSerializationSchemaBuilder.java
@@ -347,22 +347,13 @@
                                             context.getPartitionsForTopic(targetTopic)))
                             : OptionalInt.empty();
 
-            if (headerProvider != null) {
-                return new ProducerRecord<>(
-                        targetTopic,
-                        partition.isPresent() ? partition.getAsInt() : null,
-                        timestamp == null || timestamp < 0L ? null : timestamp,
-                        key,
-                        value,
-                        headerProvider.getHeaders(element));
-            } else {
-                return new ProducerRecord<>(
-                        targetTopic,
-                        partition.isPresent() ? partition.getAsInt() : null,
-                        timestamp == null || timestamp < 0L ? null : timestamp,
-                        key,
-                        value);
-            }
+            return new ProducerRecord<>(
+                    targetTopic,
+                    partition.isPresent() ? partition.getAsInt() : null,
+                    timestamp == null || timestamp < 0L ? null : timestamp,
+                    key,
+                    value,
+                    headerProvider != null ? headerProvider.getHeaders(element) : null);
         }
     }
 }