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

[loguylo]

Summary

NIFI-16097

Tracking

Please complete the following tracking steps prior to pull request creation.

Issue Tracking

• Apache NiFi Jira issue created

Pull Request Tracking

• Pull Request title starts with Apache NiFi Jira issue number, such as NIFI-00000
• Pull Request commit message starts with Apache NiFi Jira issue number, as such NIFI-00000
• Pull request contains commits signed with a registered key indicating Verified status

Pull Request Formatting

• Pull Request based on current revision of the main branch
• Pull Request refers to a feature branch with one commit containing changes

Verification

Please indicate the verification steps performed prior to pull request creation.

Build

• Build completed using ./mvnw clean install -P contrib-check
  • JDK 21
  • JDK 25

Licensing

• New dependencies are compatible with the Apache License 2.0 according to the License Policy
• New dependencies are documented in applicable LICENSE and NOTICE files

Documentation

• Documentation formatting appears as expected in rendered files

[exceptionfactory]

Thanks for proposing this additional configuration option @loguylo. This looks like a straightforward addition on initial review, I noted one stylistic recommendation.

[exceptionfactory]

Thanks @loguylo, please review the integration test failures, which may require re-evaluating the default setting of the new property.

org.apache.nifi.processor.exception.ProcessException: Failed to create Kafka Producer
	at org.apache.nifi.kafka.service.producer.Kafka3ProducerService.<init>(Kafka3ProducerService.java:61)
	at org.apache.nifi.kafka.service.Kafka3ConnectionService.getProducerService(Kafka3ConnectionService.java:298)
	at org.apache.nifi.kafka.service.Kafka3ConnectionServiceBaseIT.testProduceOneWithTransaction(Kafka3ConnectionServiceBaseIT.java:256)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)
Caused by: org.apache.kafka.common.config.ConfigException: Cannot set a transactional.id without also enabling idempotence.
	at org.apache.kafka.clients.producer.ProducerConfig.postProcessAndValidateIdempotenceConfigs(ProducerConfig.java:635)
	at org.apache.kafka.clients.producer.ProducerConfig.postProcessParsedConfig(ProducerConfig.java:574)
	at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:119)
	at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:136)
	at org.apache.kafka.clients.producer.ProducerConfig.<init>(ProducerConfig.java:692)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:313)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:340)
	at org.apache.nifi.kafka.service.producer.Kafka3ProducerService.<init>(Kafka3ProducerService.java:59)
	... 5 more

2026-07-10 14:14:43:611 [kafka-coordinator-heartbeat-thread | Kafka3ConnectionService] INFO org.apache.kafka.clients.NetworkClient - [Consumer clientId=consumer-Kafka3ConnectionService-10, groupId=Kafka3ConnectionService] Node 1 disconnected.
2026-07-10 14:14:43:612 [kafka-coordinator-heartbeat-thread | Kafka3ConnectionService] WARN org.apache.kafka.clients.NetworkClient - [Consumer clientId=consumer-Kafka3ConnectionService-10, groupId=Kafka3ConnectionService] Connection to node 1 (localhost/127.0.0.1:32925) could not be established. Node may not be available.
Error:  Errors: 
Error:    Kafka3ConnectionServiceBaseIT.testProduceOneWithTransaction:256 » Process Failed to create Kafka Producer
Error:    Kafka3ConnectionServiceSSLIT>Kafka3ConnectionServiceBaseIT.testProduceOneWithTransaction:256 » Process Failed to create Kafka Producer
Error:    Kafka3ConnectionServiceSaslPlaintextIT>Kafka3ConnectionServiceBaseIT.testProduceOneWithTransaction:256 » Process Failed to create Kafka Producer

[loguylo]

Thanks @exceptionfactory, After further research, it seems like the correct behavior is shown here. Based on the documentation, the enable.idempotence property of the kafka producer should be blocked behind the constraint that acks=all (DELIVERY_GUARANTEE=DELIVERY_REPLICATED). Furthermore, transactions should be locked behind the constraint that enable.idempotence is set to true.

It means that enable.idempotence should be set to false unless acks=all. if acks=all we can let the user choose whether he wants to set enable.idempotence to be true/false. Additionally, transactions should be turned off, unless enable.idempotence is set to true. in that case, we can let the user choose whether he wants to use transactions or not.

The current implementation simply asks the user to disable transactions if acks!=all.

I believe the structure should look like that -

ENABLE_IDEMPOTENCE = PropertyDescriptor.Builder()
    .dependsOn(DELIVERY_GUARANTEE, DeliveryGuarantee.DELIVERY_REPLICATED);
    
TRANSACTIONS_ENABLED = new PropertyDescriptor.Builder()
    .dependsOn(ENABLE_IDEMPOTENCE, "true");
    
TRANSACTIONAL_ID_PREFIX = new PropertyDescriptor.Builder()
    .dependsOn(TRANSACTIONS_ENABLED, "true");

This also means that the deafult value of ENABLE_IDEMPOTENCE and the default value of TRANSACTIONS_ENABLED should be set to false, in order to avoid conflicting configurations

What do you think?

[loguylo]

Forgot to mention, the default values of these properties should also change, so the default values of enable.idempotence and enable transactions shoud be set to false.

[exceptionfactory]

Thanks for evaluating the options @loguylo. The proposal to set enable.idempotence to false by default, and add the dependent checks on the other properties looks like a good way forward.