This is an automated email from the ASF dual-hosted git repository.
freeznet pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new c2c251a5 feat(kafka): copy Pulsar properties to Kafka headers and
support idempotence (#36)
c2c251a5 is described below
commit c2c251a5c1db68ba2024ef2d733bcccfe6e6dd04
Author: Rui Fu <[email protected]>
AuthorDate: Sat Jul 11 16:50:33 2026 +0800
feat(kafka): copy Pulsar properties to Kafka headers and support
idempotence (#36)
Co-authored-by: David Kjerrumgaard
<[email protected]>
---
.../apache/pulsar/io/kafka/KafkaAbstractSink.java | 45 ++++-
.../apache/pulsar/io/kafka/KafkaSinkConfig.java | 21 +-
.../io/kafka/sink/KafkaAbstractSinkTest.java | 213 +++++++++++++++++++++
3 files changed, 272 insertions(+), 7 deletions(-)
diff --git
a/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java
b/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java
index 232fd7ff..dfc40c54 100644
--- a/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java
+++ b/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java
@@ -19,9 +19,14 @@
package org.apache.pulsar.io.kafka;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
@@ -48,13 +53,10 @@ public abstract class KafkaAbstractSink<K, V> implements
Sink<byte[]> {
@Override
public void write(Record<byte[]> sourceRecord) {
- KeyValue<K, V> keyValue = extractKeyValue(sourceRecord);
- ProducerRecord<K, V> record = new
ProducerRecord<>(kafkaSinkConfig.getTopic(), keyValue.getKey(),
- keyValue.getValue());
+ ProducerRecord<K, V> record = buildProducerRecord(sourceRecord);
if (log.isDebugEnabled()) {
log.debug("Record sending to kafka, record={}.", record);
}
-
producer.send(record, (metadata, exception) -> {
if (exception == null) {
sourceRecord.ack();
@@ -64,6 +66,23 @@ public abstract class KafkaAbstractSink<K, V> implements
Sink<byte[]> {
});
}
+ // protected (NOT package-private) so the test subclass can expose it;
package-private would be invisible
+ // because the test lives in package org.apache.pulsar.io.kafka.sink.
+ protected ProducerRecord<K, V> buildProducerRecord(Record<byte[]>
sourceRecord) {
+ KeyValue<K, V> keyValue = extractKeyValue(sourceRecord);
+ Map<String, String> properties = sourceRecord.getProperties();
+ if (!kafkaSinkConfig.isCopyHeadersEnabled() || properties == null ||
properties.isEmpty()) {
+ return new ProducerRecord<>(kafkaSinkConfig.getTopic(),
keyValue.getKey(), keyValue.getValue());
+ }
+ List<Header> headers = new ArrayList<>(properties.size());
+ for (Map.Entry<String, String> entry : properties.entrySet()) {
+ if (entry.getKey() != null && entry.getValue() != null) {
+ headers.add(new RecordHeader(entry.getKey(),
entry.getValue().getBytes(StandardCharsets.UTF_8)));
+ }
+ }
+ return new ProducerRecord<>(kafkaSinkConfig.getTopic(), null,
keyValue.getKey(), keyValue.getValue(), headers);
+ }
+
@Override
public void close() throws IOException {
if (producer != null) {
@@ -114,7 +133,23 @@ public abstract class KafkaAbstractSink<K, V> implements
Sink<byte[]> {
if
(StringUtils.isNotEmpty(kafkaSinkConfig.getSslTruststorePassword())) {
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,
kafkaSinkConfig.getSslTruststorePassword());
}
- props.put(ProducerConfig.ACKS_CONFIG, kafkaSinkConfig.getAcks());
+ // If the user has enabled idempotent production (via
producerConfigProperties), Kafka
+ // requires acks=all. Do not silently downgrade acks in that case;
instead enforce all.
+ // Without idempotence the configured acks value is applied as before.
+ boolean idempotenceEnabled = Boolean.parseBoolean(
+
String.valueOf(props.getOrDefault(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,
"false")));
+ if (idempotenceEnabled) {
+ Object rawAcks = props.get(ProducerConfig.ACKS_CONFIG);
+ String userAcks = rawAcks != null ? String.valueOf(rawAcks) : null;
+ if (userAcks != null && !userAcks.equals("all") &&
!userAcks.equals("-1")) {
+ throw new IllegalArgumentException(
+ "enable.idempotence=true requires acks=all, but acks
was set to: " + userAcks);
+ }
+ // Enforce acks=all when idempotence is enabled; ignore the
top-level acks field.
+ props.put(ProducerConfig.ACKS_CONFIG, "all");
+ } else {
+ props.put(ProducerConfig.ACKS_CONFIG, kafkaSinkConfig.getAcks());
+ }
props.put(ProducerConfig.BATCH_SIZE_CONFIG,
String.valueOf(kafkaSinkConfig.getBatchSize()));
props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,
String.valueOf(kafkaSinkConfig.getMaxRequestSize()));
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
kafkaSinkConfig.getKeySerializerClass());
diff --git
a/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaSinkConfig.java
b/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaSinkConfig.java
index b63e3756..ba716273 100644
--- a/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaSinkConfig.java
+++ b/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaSinkConfig.java
@@ -113,11 +113,28 @@ public class KafkaSinkConfig implements Serializable {
"The serializer class for Kafka producer to serialize
values. You typically shouldn't care this. "
+ "Since the serializer will be set by a specific
implementation of `KafkaAbstractSink`.")
private String valueSerializerClass =
"org.apache.kafka.common.serialization.ByteArraySerializer";
+ @FieldDoc(
+ required = false,
+ defaultValue = "false",
+ help = "If true, the Pulsar message properties are copied into the
Kafka record headers (each property "
+ + "name becomes a header key, the value is UTF-8 encoded).
Useful with the Kinesis source "
+ + "messageKeyMode=SHARD_ID: the Kafka message key carries
the shardId for ordered routing while the "
+ + "original partition key is preserved in the
'kinesis.partition.key' header for consumers.")
+ private boolean copyHeadersEnabled = false;
+
@FieldDoc(
defaultValue = "",
help =
- "The producer config properties to be passed to Producer.
Note that other properties specified "
- + "in the connector config file take precedence
over this config.")
+ "The producer config properties to be passed to Producer.
Note that most properties specified "
+ + "in the connector config file (bootstrapServers,
acks, batchSize, etc.) take precedence "
+ + "over entries in this map. Exception: if
'enable.idempotence=true' is set here, the "
+ + "sink will enforce acks=all and ignore the
top-level acks field, because Kafka's "
+ + "idempotent producer requires acks=all. "
+ + "To preserve per-shard ordering when routing
Kinesis records to Kafka, set "
+ + "'enable.idempotence=true' and
'max.in.flight.requests.per.connection=1' (or <=5 for "
+ + "Kafka >= 1.0.0 with idempotence) in this map,
and configure the Kinesis source with "
+ + "messageKeyMode=SHARD_ID so that all records
from the same shard share a key and land "
+ + "on the same Kafka partition in arrival order.")
private Map<String, Object> producerConfigProperties;
public static KafkaSinkConfig load(String yamlFile) throws IOException {
diff --git
a/kafka/src/test/java/org/apache/pulsar/io/kafka/sink/KafkaAbstractSinkTest.java
b/kafka/src/test/java/org/apache/pulsar/io/kafka/sink/KafkaAbstractSinkTest.java
index f8c5b1df..b092b8ea 100644
---
a/kafka/src/test/java/org/apache/pulsar/io/kafka/sink/KafkaAbstractSinkTest.java
+++
b/kafka/src/test/java/org/apache/pulsar/io/kafka/sink/KafkaAbstractSinkTest.java
@@ -24,16 +24,21 @@ import static org.testng.Assert.fail;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.common.io.SinkConfig;
@@ -55,6 +60,141 @@ public class KafkaAbstractSinkTest {
}
}
+ /**
+ * A DummySink variant that captures the final producer props (after all
open() merging logic)
+ * so tests can assert on the resolved configuration without needing a
live broker.
+ */
+ private static class CapturingSink extends KafkaAbstractSink<String,
byte[]> {
+ Properties capturedProps;
+
+ @Override
+ public KeyValue<String, byte[]> extractKeyValue(Record<byte[]> record)
{
+ return new KeyValue<>(record.getKey().orElse(null),
record.getValue());
+ }
+
+ @Override
+ protected Properties beforeCreateProducer(Properties props) {
+ capturedProps = new Properties();
+ capturedProps.putAll(props);
+ return props;
+ }
+ }
+
+ /**
+ * DummySink variant that exposes buildProducerRecord() for direct unit
testing, so tests can
+ * assert on header propagation logic without needing a live Kafka broker.
+ */
+ private static class HeaderCapturingSink extends KafkaAbstractSink<String,
byte[]> {
+
+ @Override
+ public KeyValue<String, byte[]> extractKeyValue(Record<byte[]> record)
{
+ return new KeyValue<>(record.getKey().orElse(null),
record.getValue());
+ }
+
+ public ProducerRecord<String, byte[]>
buildRecordForTest(Record<byte[]> r) {
+ return buildProducerRecord(r);
+ }
+ }
+
+ /** Creates an anonymous Record<byte[]> with the given key, value, and
properties. */
+ private static Record<byte[]> makeRecord(String key, byte[] value,
Map<String, String> properties) {
+ return new Record<byte[]>() {
+ @Override
+ public Optional<String> getKey() {
+ return Optional.ofNullable(key);
+ }
+
+ @Override
+ public byte[] getValue() {
+ return value;
+ }
+
+ @Override
+ public Map<String, String> getProperties() {
+ return properties;
+ }
+ };
+ }
+
+ @Test
+ public void testCopyHeadersDisabledByDefaultProducesNoHeaders() throws
Exception {
+ HeaderCapturingSink sink = new HeaderCapturingSink();
+ Map<String, Object> config = validConfig();
+ // copyHeadersEnabled is NOT set — default false
+ sink.open(config, makeSinkContext());
+ try {
+ Map<String, String> props = new HashMap<>();
+ props.put("kinesis.partition.key", "userA");
+ Record<byte[]> record = makeRecord("shard-1",
"payload".getBytes(StandardCharsets.UTF_8), props);
+
+ ProducerRecord<String, byte[]> produced =
sink.buildRecordForTest(record);
+
+ assertEquals(produced.key(), "shard-1");
+ // No headers should be present when copyHeadersEnabled=false
+ int headerCount = 0;
+ for (Header ignored : produced.headers()) {
+ headerCount++;
+ }
+ assertEquals(headerCount, 0, "Expected no headers when
copyHeadersEnabled=false");
+ } finally {
+ sink.close();
+ }
+ }
+
+ @Test
+ public void testCopyHeadersEnabledPropagatesPropertiesAsHeaders() throws
Exception {
+ HeaderCapturingSink sink = new HeaderCapturingSink();
+ Map<String, Object> config = validConfig();
+ config.put("copyHeadersEnabled", true);
+ sink.open(config, makeSinkContext());
+ try {
+ Map<String, String> props = new HashMap<>();
+ props.put("kinesis.partition.key", "userA");
+ props.put("kinesis.shard.id", "shard-1");
+ Record<byte[]> record = makeRecord("shard-1",
"payload".getBytes(StandardCharsets.UTF_8), props);
+
+ ProducerRecord<String, byte[]> produced =
sink.buildRecordForTest(record);
+
+ assertEquals(produced.key(), "shard-1");
+
+ // Collect produced headers into a map for easy assertion
+ Map<String, byte[]> headerMap = new HashMap<>();
+ for (Header h : produced.headers()) {
+ headerMap.put(h.key(), h.value());
+ }
+
+ assertEquals(headerMap.size(), 2, "Expected exactly 2 headers");
+ assertEquals(
+ new String(headerMap.get("kinesis.partition.key"),
StandardCharsets.UTF_8),
+ "userA",
+ "kinesis.partition.key header must carry the original
partition key");
+ } finally {
+ sink.close();
+ }
+ }
+
+ @Test
+ public void testCopyHeadersEnabledWithEmptyPropertiesProducesNoHeaders()
throws Exception {
+ HeaderCapturingSink sink = new HeaderCapturingSink();
+ Map<String, Object> config = validConfig();
+ config.put("copyHeadersEnabled", true);
+ sink.open(config, makeSinkContext());
+ try {
+ Record<byte[]> record = makeRecord("key1",
"data".getBytes(StandardCharsets.UTF_8),
+ Collections.emptyMap());
+
+ ProducerRecord<String, byte[]> produced =
sink.buildRecordForTest(record);
+
+ int headerCount = 0;
+ for (Header ignored : produced.headers()) {
+ headerCount++;
+ }
+ assertEquals(headerCount, 0, "Expected no headers when properties
map is empty");
+ } finally {
+ sink.close();
+ }
+ }
+
@FunctionalInterface
public interface ThrowingRunnable {
void run() throws Throwable;
@@ -305,6 +445,79 @@ public class KafkaAbstractSinkTest {
assertEquals(config.getSslTruststorePassword(), "cert_pwd");
}
+ /**
+ * Verifies that setting enable.idempotence=true in
producerConfigProperties results in
+ * acks=all in the final producer configuration (the top-level acks field
must NOT clobber it).
+ */
+ @Test
+ public void testIdempotenceEnabledSetsAcksAll() throws Exception {
+ CapturingSink sink = new CapturingSink();
+ Map<String, Object> config = validConfig();
+ // acks=1 at top level; idempotence=true in producerConfigProperties
+ Map<String, Object> extraProps = new HashMap<>();
+ extraProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
+ config.put("producerConfigProperties", extraProps);
+
+ sink.open(config, makeSinkContext());
+ sink.close();
+
+ assertEquals("all",
sink.capturedProps.getProperty(ProducerConfig.ACKS_CONFIG),
+ "acks must be 'all' when enable.idempotence=true");
+ }
+
+ /**
+ * Verifies that setting enable.idempotence=true together with an explicit
conflicting acks
+ * value in producerConfigProperties yields a clear
IllegalArgumentException rather than a
+ * cryptic Kafka ConfigException.
+ */
+ @Test
+ public void testIdempotenceWithConflictingAcksThrows() throws Exception {
+ CapturingSink sink = new CapturingSink();
+ Map<String, Object> config = validConfig();
+ Map<String, Object> extraProps = new HashMap<>();
+ extraProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
+ extraProps.put(ProducerConfig.ACKS_CONFIG, "1");
+ config.put("producerConfigProperties", extraProps);
+
+ expectThrows(IllegalArgumentException.class,
+ "enable.idempotence=true requires acks=all, but acks was set
to: 1",
+ () -> {
+ try {
+ sink.open(config, makeSinkContext());
+ } finally {
+ sink.close();
+ }
+ });
+ }
+
+ /** Minimal no-op SinkContext for use in unit tests. */
+ private static SinkContext makeSinkContext() {
+ return new SinkContext() {
+ @Override public int getInstanceId() { return 0; }
+ @Override public int getNumInstances() { return 0; }
+ @Override public void recordMetric(String metricName, double
value) {}
+ @Override public Collection<String> getInputTopics() { return
null; }
+ @Override public SinkConfig getSinkConfig() { return null; }
+ @Override public String getTenant() { return null; }
+ @Override public String getNamespace() { return null; }
+ @Override public String getSinkName() { return null; }
+ @Override public Logger getLogger() { return null; }
+ @Override public String getSecret(String key) { return null; }
+ @Override public void incrCounter(String key, long amount) {}
+ @Override public CompletableFuture<Void> incrCounterAsync(String
key, long amount) { return null; }
+ @Override public long getCounter(String key) { return 0; }
+ @Override public CompletableFuture<Long> getCounterAsync(String
key) { return null; }
+ @Override public void putState(String key, ByteBuffer value) {}
+ @Override public CompletableFuture<Void> putStateAsync(String key,
ByteBuffer value) { return null; }
+ @Override public ByteBuffer getState(String key) { return null; }
+ @Override public CompletableFuture<ByteBuffer>
getStateAsync(String key) { return null; }
+ @Override public void deleteState(String key) {}
+ @Override public CompletableFuture<Void> deleteStateAsync(String
key) { return null; }
+ @Override public PulsarClient getPulsarClient() { return null; }
+ @Override public void fatal(Throwable t) {}
+ };
+ }
+
private File getFile(String name) {
ClassLoader classLoader = getClass().getClassLoader();
return new File(classLoader.getResource(name).getFile());
