Skip to content

NIFI-16097 Added enable.idemotence Option in Publish Kafka Processor #11413

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
loguylo wants to merge 1 commit into
base: main
Choose a base branch
from
+31 −7
Open

NIFI-16097 Added enable.idemotence Option in Publish Kafka Processor #11413

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 18 additions & 2 deletions ...le/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/PublishKafka.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,19 @@ public class PublishKafka extends AbstractProcessor implements VerifiableProcess
.defaultValue(DeliveryGuarantee.DELIVERY_REPLICATED)
.build();

static final PropertyDescriptor ENABLE_IDEMPOTENCE = new PropertyDescriptor.Builder()
.name("enable.idempotence")
.displayName("Enable Idempotence")
.description("Specifies whether the producer will ensure that exactly one copy of each message is written in the stream. " +
"If set to ‘false’, producer retries due to broker failures, etc., may write duplicates of the retried message in the stream. " +
"Corresponds to Kafka Client enable.idempotence property.")
Comment on lines +146 to +148

@exceptionfactory exceptionfactory Jul 10, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A multiline string can be used instead of string concatenation for the description

.expressionLanguageSupported(ExpressionLanguageScope.NONE)
.dependsOn(DELIVERY_GUARANTEE, DeliveryGuarantee.DELIVERY_REPLICATED)
.required(false)
.allowableValues("true", "false")
.defaultValue("true")
.build();

static final PropertyDescriptor COMPRESSION_CODEC = new PropertyDescriptor.Builder()
.name("compression.type")
.displayName("Compression Type")
Expand Down Expand Up @@ -312,6 +325,7 @@ public class PublishKafka extends AbstractProcessor implements VerifiableProcess
TOPIC_NAME,
FAILURE_STRATEGY,
DELIVERY_GUARANTEE,
ENABLE_IDEMPOTENCE,
COMPRESSION_CODEC,
MAX_REQUEST_SIZE,
TRANSACTIONS_ENABLED,
Expand Down Expand Up @@ -361,11 +375,12 @@ public List<ConfigVerificationResult> verify(final ProcessContext context, final
final String transactionalIdPrefix = transactionsEnabled ? context.getProperty(TRANSACTIONAL_ID_PREFIX).evaluateAttributeExpressions().getValue() : null;
final Supplier<String> transactionalIdSupplier = new TransactionIdSupplier(transactionalIdPrefix);
final String deliveryGuarantee = context.getProperty(DELIVERY_GUARANTEE).getValue();
final boolean enableIdempotence = context.getProperty(ENABLE_IDEMPOTENCE).asBoolean();
final String compressionCodec = context.getProperty(COMPRESSION_CODEC).getValue();
final String partitionClass = context.getProperty(PARTITION_CLASS).getValue();
final int maxRequestSize = context.getProperty(MAX_REQUEST_SIZE).asDataSize(DataUnit.B).intValue();
final ProducerConfiguration producerConfiguration = new ProducerConfiguration(
transactionsEnabled, transactionalIdSupplier.get(), deliveryGuarantee, compressionCodec, partitionClass, maxRequestSize);
transactionsEnabled, transactionalIdSupplier.get(), deliveryGuarantee, enableIdempotence, compressionCodec, partitionClass, maxRequestSize);

try (final KafkaProducerService producerService = connectionService.getProducerService(producerConfiguration)) {
final ConfigVerificationResult.Builder verificationPartitions = new ConfigVerificationResult.Builder()
Expand Down Expand Up @@ -454,12 +469,13 @@ private KafkaProducerService createProducerService(final ProcessContext context)
final boolean transactionsEnabled = context.getProperty(TRANSACTIONS_ENABLED).asBoolean();
final String transactionalIdPrefix = transactionsEnabled ? context.getProperty(TRANSACTIONAL_ID_PREFIX).evaluateAttributeExpressions().getValue() : null;
final String deliveryGuarantee = context.getProperty(DELIVERY_GUARANTEE).getValue();
final boolean enableIdempotence = context.getProperty(ENABLE_IDEMPOTENCE).asBoolean();
final String compressionCodec = context.getProperty(COMPRESSION_CODEC).getValue();
final String partitionClass = context.getProperty(PARTITION_CLASS).getValue();
final int maxRequestSize = context.getProperty(MAX_REQUEST_SIZE).asDataSize(DataUnit.B).intValue();

final ProducerConfiguration producerConfiguration = new ProducerConfiguration(
transactionsEnabled, transactionalIdPrefix, deliveryGuarantee, compressionCodec, partitionClass, maxRequestSize);
transactionsEnabled, transactionalIdPrefix, deliveryGuarantee, enableIdempotence, compressionCodec, partitionClass, maxRequestSize);

return connectionService.getProducerService(producerConfiguration);
}
Expand Down
7 changes: 7 additions & 0 deletions ...e-api/src/main/java/org/apache/nifi/kafka/service/api/producer/ProducerConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,19 +20,22 @@ public class ProducerConfiguration {
private final boolean transactionsEnabled;
private final String transactionIdPrefix;
private final String deliveryGuarantee;
private final boolean enableIdempotence;
private final String compressionCodec;
private final String partitionClass;
private final int maxRequestSize;

public ProducerConfiguration(final boolean transactionsEnabled,
final String transactionIdPrefix,
final String deliveryGuarantee,
final boolean enableIdempotence,
final String compressionCodec,
final String partitionClass,
final int maxRequestSize) {
this.transactionsEnabled = transactionsEnabled;
this.transactionIdPrefix = transactionIdPrefix;
this.deliveryGuarantee = deliveryGuarantee;
this.enableIdempotence = enableIdempotence;
this.compressionCodec = compressionCodec;
this.partitionClass = partitionClass;
this.maxRequestSize = maxRequestSize;
Expand All @@ -50,6 +53,10 @@ public String getDeliveryGuarantee() {
return deliveryGuarantee;
}

public boolean getEnableIdempotence() {
return enableIdempotence;
}

public String getCompressionCodec() {
return compressionCodec;
}
Expand Down
1 change: 1 addition & 0 deletions ...a-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,6 +281,7 @@ public KafkaProducerService getProducerService(final ProducerConfiguration produ
if (producerConfiguration.getDeliveryGuarantee() != null) {
properties.put(ProducerConfig.ACKS_CONFIG, producerConfiguration.getDeliveryGuarantee());
}
properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, producerConfiguration.getEnableIdempotence());
if (producerConfiguration.getCompressionCodec() != null) {
properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, producerConfiguration.getCompressionCodec());
}
Expand Down
10 changes: 5 additions & 5 deletions ...ice-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceBaseIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,7 +241,7 @@ void testAdminClient() throws ExecutionException, InterruptedException, TimeoutE

@Test
void testProduceOneNoTransaction() {
final ProducerConfiguration producerConfiguration = new ProducerConfiguration(false, null, null, null, null, 1_000_000);
final ProducerConfiguration producerConfiguration = new ProducerConfiguration(false, null, null, false, null, null, 1_000_000);
final KafkaProducerService producerService = service.getProducerService(producerConfiguration);
final KafkaRecord kafkaRecord = new KafkaRecord(null, null, null, null, RECORD_VALUE, emptyList());
final List<KafkaRecord> kafkaRecords = Collections.singletonList(kafkaRecord);
Expand All @@ -252,7 +252,7 @@ void testProduceOneNoTransaction() {

@Test
void testProduceOneWithTransaction() {
final ProducerConfiguration producerConfiguration = new ProducerConfiguration(true, "transaction-", null, null, null, 1_000_000);
final ProducerConfiguration producerConfiguration = new ProducerConfiguration(true, "transaction-", null, false, null, null, 1_000_000);
final KafkaProducerService producerService = service.getProducerService(producerConfiguration);
final KafkaRecord kafkaRecord = new KafkaRecord(null, null, null, null, RECORD_VALUE, emptyList());
final List<KafkaRecord> kafkaRecords = Collections.singletonList(kafkaRecord);
Expand All @@ -263,7 +263,7 @@ void testProduceOneWithTransaction() {

@Test
void testProduceConsumeRecord() {
final ProducerConfiguration producerConfiguration = new ProducerConfiguration(false, null, null, null, null, 1_000_000);
final ProducerConfiguration producerConfiguration = new ProducerConfiguration(false, null, null, false, null, null, 1_000_000);
final KafkaProducerService producerService = service.getProducerService(producerConfiguration);

final long timestamp = System.currentTimeMillis();
Expand Down Expand Up @@ -300,7 +300,7 @@ void testCurrentLag() {
final String groupId = "Group_" + timestamp;
final String topic = "Topic_" + timestamp;
final int partition = 0;
final ProducerConfiguration producerConfiguration = new ProducerConfiguration(false, null, null, null, null, 1_000_000);
final ProducerConfiguration producerConfiguration = new ProducerConfiguration(false, null, null, false, null, null, 1_000_000);
final KafkaProducerService producerService = service.getProducerService(producerConfiguration);

final KafkaRecord kafkaRecord = new KafkaRecord(null, partition, timestamp, RECORD_KEY, RECORD_VALUE, emptyList());
Expand Down Expand Up @@ -404,7 +404,7 @@ private void assertResultFound(

@Test
void testGetProducerService() {
final ProducerConfiguration producerConfiguration = new ProducerConfiguration(false, null, null, null, null, 1_000_000);
final ProducerConfiguration producerConfiguration = new ProducerConfiguration(false, null, null, false, null, null, 1_000_000);
final KafkaProducerService producerService = service.getProducerService(producerConfiguration);
final List<PartitionState> partitionStates = producerService.getPartitionStates(TOPIC);
assertPartitionStatesFound(partitionStates);
Expand Down
Loading