From 2002bb121b9ed51b2692c68280ffc3b3e2a4f51d Mon Sep 17 00:00:00 2001
From: Pavel Kotelevsky <38818382+chillleader@users.noreply.github.com>
Date: Thu, 25 Apr 2024 14:37:19 +0200
Subject: [PATCH] feat(kafka,sqs): support inbound connector deduplication
 (#2340)

---
 .../aws-sqs-boundary-connector.json           | 64 +++++++++++++++++++
 ...ws-sqs-inbound-intermediate-connector.json | 64 +++++++++++++++++++
 .../aws-sqs-start-event-connector.json        | 64 +++++++++++++++++++
 .../aws-sqs-start-message.json                | 64 +++++++++++++++++++
 connectors/aws/aws-sqs/pom.xml                |  3 +
 ...fka-inbound-connector-boundary-hybrid.json | 64 +++++++++++++++++++
 ...inbound-connector-intermediate-hybrid.json | 64 +++++++++++++++++++
 ...-inbound-connector-start-event-hybrid.json | 64 +++++++++++++++++++
 ...nbound-connector-start-message-hybrid.json | 64 +++++++++++++++++++
 .../kafka-inbound-connector-boundary.json     | 64 +++++++++++++++++++
 .../kafka-inbound-connector-intermediate.json | 64 +++++++++++++++++++
 .../kafka-inbound-connector-start-event.json  | 64 +++++++++++++++++++
 ...kafka-inbound-connector-start-message.json | 64 +++++++++++++++++++
 connectors/kafka/pom.xml                      |  3 +
 14 files changed, 774 insertions(+)

diff --git a/connectors/aws/aws-sqs/element-templates/aws-sqs-boundary-connector.json b/connectors/aws/aws-sqs/element-templates/aws-sqs-boundary-connector.json
index 920588e153..393154d8b1 100644
--- a/connectors/aws/aws-sqs/element-templates/aws-sqs-boundary-connector.json
+++ b/connectors/aws/aws-sqs/element-templates/aws-sqs-boundary-connector.json
@@ -35,6 +35,10 @@
   }, {
     "id" : "correlation",
     "label" : "Correlation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -246,6 +250,66 @@
       "type" : "bpmn:Message#property"
     },
     "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/aws/aws-sqs/element-templates/aws-sqs-inbound-intermediate-connector.json b/connectors/aws/aws-sqs/element-templates/aws-sqs-inbound-intermediate-connector.json
index 89d298a605..d0a047c96b 100644
--- a/connectors/aws/aws-sqs/element-templates/aws-sqs-inbound-intermediate-connector.json
+++ b/connectors/aws/aws-sqs/element-templates/aws-sqs-inbound-intermediate-connector.json
@@ -35,6 +35,10 @@
   }, {
     "id" : "correlation",
     "label" : "Correlation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -246,6 +250,66 @@
       "type" : "bpmn:Message#property"
     },
     "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/aws/aws-sqs/element-templates/aws-sqs-start-event-connector.json b/connectors/aws/aws-sqs/element-templates/aws-sqs-start-event-connector.json
index 7e849e54c3..a4258369e8 100644
--- a/connectors/aws/aws-sqs/element-templates/aws-sqs-start-event-connector.json
+++ b/connectors/aws/aws-sqs/element-templates/aws-sqs-start-event-connector.json
@@ -31,6 +31,10 @@
   }, {
     "id" : "activation",
     "label" : "Activation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -191,6 +195,66 @@
       "type" : "zeebe:property"
     },
     "type" : "String"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/aws/aws-sqs/element-templates/aws-sqs-start-message.json b/connectors/aws/aws-sqs/element-templates/aws-sqs-start-message.json
index 33d557dc5c..ecc04c72f4 100644
--- a/connectors/aws/aws-sqs/element-templates/aws-sqs-start-message.json
+++ b/connectors/aws/aws-sqs/element-templates/aws-sqs-start-message.json
@@ -35,6 +35,10 @@
   }, {
     "id" : "correlation",
     "label" : "Correlation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -274,6 +278,66 @@
       "type" : "bpmn:Message#property"
     },
     "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/aws/aws-sqs/pom.xml b/connectors/aws/aws-sqs/pom.xml
index 274e968d66..a4170297d2 100644
--- a/connectors/aws/aws-sqs/pom.xml
+++ b/connectors/aws/aws-sqs/pom.xml
@@ -75,6 +75,9 @@
                   <templateFileName>aws-sqs-boundary-connector.json</templateFileName>
                 </file>
               </files>
+              <features>
+                <INBOUND_DEDUPLICATION>true</INBOUND_DEDUPLICATION>
+              </features>
             </connector>
           </connectors>
           <includeDependencies>
diff --git a/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-boundary-hybrid.json b/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-boundary-hybrid.json
index 943a3f9231..066aaa856d 100644
--- a/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-boundary-hybrid.json
+++ b/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-boundary-hybrid.json
@@ -32,6 +32,10 @@
   }, {
     "id" : "correlation",
     "label" : "Correlation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -279,6 +283,66 @@
       "type" : "bpmn:Message#property"
     },
     "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-intermediate-hybrid.json b/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-intermediate-hybrid.json
index 0491b9532d..98cb917f45 100644
--- a/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-intermediate-hybrid.json
+++ b/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-intermediate-hybrid.json
@@ -32,6 +32,10 @@
   }, {
     "id" : "correlation",
     "label" : "Correlation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -279,6 +283,66 @@
       "type" : "bpmn:Message#property"
     },
     "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-start-event-hybrid.json b/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-start-event-hybrid.json
index 0b95f24a24..5e05261dbf 100644
--- a/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-start-event-hybrid.json
+++ b/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-start-event-hybrid.json
@@ -28,6 +28,10 @@
   }, {
     "id" : "activation",
     "label" : "Activation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -224,6 +228,66 @@
       "type" : "zeebe:property"
     },
     "type" : "String"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-start-message-hybrid.json b/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-start-message-hybrid.json
index 46a35519a0..596a838d04 100644
--- a/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-start-message-hybrid.json
+++ b/connectors/kafka/element-templates/hybrid/kafka-inbound-connector-start-message-hybrid.json
@@ -32,6 +32,10 @@
   }, {
     "id" : "correlation",
     "label" : "Correlation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -307,6 +311,66 @@
       "type" : "bpmn:Message#property"
     },
     "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/kafka/element-templates/kafka-inbound-connector-boundary.json b/connectors/kafka/element-templates/kafka-inbound-connector-boundary.json
index 5ec13a6aad..88e7e7810e 100644
--- a/connectors/kafka/element-templates/kafka-inbound-connector-boundary.json
+++ b/connectors/kafka/element-templates/kafka-inbound-connector-boundary.json
@@ -29,6 +29,10 @@
   }, {
     "id" : "correlation",
     "label" : "Correlation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -274,6 +278,66 @@
       "type" : "bpmn:Message#property"
     },
     "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/kafka/element-templates/kafka-inbound-connector-intermediate.json b/connectors/kafka/element-templates/kafka-inbound-connector-intermediate.json
index 80919c2f20..dcb62b1628 100644
--- a/connectors/kafka/element-templates/kafka-inbound-connector-intermediate.json
+++ b/connectors/kafka/element-templates/kafka-inbound-connector-intermediate.json
@@ -29,6 +29,10 @@
   }, {
     "id" : "correlation",
     "label" : "Correlation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -274,6 +278,66 @@
       "type" : "bpmn:Message#property"
     },
     "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/kafka/element-templates/kafka-inbound-connector-start-event.json b/connectors/kafka/element-templates/kafka-inbound-connector-start-event.json
index fdeb6437b0..abc72f469d 100644
--- a/connectors/kafka/element-templates/kafka-inbound-connector-start-event.json
+++ b/connectors/kafka/element-templates/kafka-inbound-connector-start-event.json
@@ -25,6 +25,10 @@
   }, {
     "id" : "activation",
     "label" : "Activation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -219,6 +223,66 @@
       "type" : "zeebe:property"
     },
     "type" : "String"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/kafka/element-templates/kafka-inbound-connector-start-message.json b/connectors/kafka/element-templates/kafka-inbound-connector-start-message.json
index 75f1358ffe..6633c2220c 100644
--- a/connectors/kafka/element-templates/kafka-inbound-connector-start-message.json
+++ b/connectors/kafka/element-templates/kafka-inbound-connector-start-message.json
@@ -29,6 +29,10 @@
   }, {
     "id" : "correlation",
     "label" : "Correlation"
+  }, {
+    "id" : "deduplication",
+    "label" : "Deduplication",
+    "tooltip" : "Deduplication allows you to configure multiple inbound connector elements to reuse the same backend (consumer/thread/endpoint) by sharing the same deduplication ID."
   }, {
     "id" : "output",
     "label" : "Output mapping"
@@ -302,6 +306,66 @@
       "type" : "bpmn:Message#property"
     },
     "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeManualFlag",
+    "label" : "Manual mode",
+    "description" : "By default, similar connectors receive the same deduplication ID. Customize by activating manual mode",
+    "value" : false,
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationModeManualFlag",
+      "type" : "zeebe:property"
+    },
+    "type" : "Boolean"
+  }, {
+    "id" : "deduplicationId",
+    "label" : "Deduplication ID",
+    "constraints" : {
+      "notEmpty" : true,
+      "pattern" : {
+        "value" : "^[a-zA-Z0-9_-]+$",
+        "message" : "Only alphanumeric characters, dashes, and underscores are allowed"
+      }
+    },
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationId",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationModeManualFlag",
+      "equals" : true,
+      "type" : "simple"
+    },
+    "type" : "String"
+  }, {
+    "id" : "deduplicationModeManual",
+    "value" : "MANUAL",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : true,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
+  }, {
+    "id" : "deduplicationModeAuto",
+    "value" : "AUTO",
+    "group" : "deduplication",
+    "binding" : {
+      "name" : "deduplicationMode",
+      "type" : "zeebe:property"
+    },
+    "condition" : {
+      "property" : "deduplicationId",
+      "isActive" : false,
+      "type" : "simple"
+    },
+    "type" : "Hidden"
   }, {
     "id" : "resultVariable",
     "label" : "Result variable",
diff --git a/connectors/kafka/pom.xml b/connectors/kafka/pom.xml
index 43a386c15e..6dc664b60c 100644
--- a/connectors/kafka/pom.xml
+++ b/connectors/kafka/pom.xml
@@ -118,6 +118,9 @@ except in compliance with the proprietary license.</license.inlineheader>
                 </file>
               </files>
               <generateHybridTemplates>true</generateHybridTemplates>
+              <features>
+                <INBOUND_DEDUPLICATION>true</INBOUND_DEDUPLICATION>
+              </features>
             </connector>
           </connectors>
         </configuration>