diff --git a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/PublishKafka.java b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/PublishKafka.java
index f68d09836f9b..3aff7b0616b4 100644
--- a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/PublishKafka.java
+++ b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-processors/src/main/java/org/apache/nifi/kafka/processors/PublishKafka.java
@@ -140,6 +140,20 @@ 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.""")
+            .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")
@@ -312,6 +326,7 @@ public class PublishKafka extends AbstractProcessor implements VerifiableProcess
             TOPIC_NAME,
             FAILURE_STRATEGY,
             DELIVERY_GUARANTEE,
+            ENABLE_IDEMPOTENCE,
             COMPRESSION_CODEC,
             MAX_REQUEST_SIZE,
             TRANSACTIONS_ENABLED,
@@ -361,11 +376,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()
@@ -454,12 +470,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);
     }
diff --git a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-api/src/main/java/org/apache/nifi/kafka/service/api/producer/ProducerConfiguration.java b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-api/src/main/java/org/apache/nifi/kafka/service/api/producer/ProducerConfiguration.java
index 4494b3846820..6dd3c265b31f 100644
--- a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-api/src/main/java/org/apache/nifi/kafka/service/api/producer/ProducerConfiguration.java
+++ b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-api/src/main/java/org/apache/nifi/kafka/service/api/producer/ProducerConfiguration.java
@@ -20,6 +20,7 @@ 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;
@@ -27,12 +28,14 @@ public class ProducerConfiguration {
     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;
@@ -50,6 +53,10 @@ public String getDeliveryGuarantee() {
         return deliveryGuarantee;
     }
 
+    public boolean getEnableIdempotence() {
+        return enableIdempotence;
+    }
+
     public String getCompressionCodec() {
         return compressionCodec;
     }
diff --git a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
index 225a21ca0682..17b7edc066cd 100644
--- a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
+++ b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/main/java/org/apache/nifi/kafka/service/Kafka3ConnectionService.java
@@ -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());
         }
diff --git a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceBaseIT.java b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceBaseIT.java
index 2960aa4bcb94..6f600bf54620 100644
--- a/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceBaseIT.java
+++ b/nifi-extension-bundles/nifi-kafka-bundle/nifi-kafka-service-shared/src/test/java/org/apache/nifi/kafka/service/Kafka3ConnectionServiceBaseIT.java
@@ -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);
@@ -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);
@@ -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();
@@ -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());
@@ -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);