merge: #10697

10697: Export missing records to ES r=remcowesterhoud a=lzgabel

## Description

Currently, the Zeebe Elasticsearch exporter doesn't allow to export all records. The following records are missing:

* `timer`
* `message_start_event_subscription`
* `process_event`
* `deployment_distribution`

## Related issues

closes #8337

## Definition of Done

Code changes:
* [ ] The changes are backwards compatibility with previous versions
* [ ] If it fixes a bug then PRs are created to [backport](https://github.com/camunda/zeebe/compare/stable/0.24...main?expand=1&template=backport_template.md&title=[Backport%200.24]) the fix to the last two minor versions. You can trigger a backport by assigning labels (e.g. `backport stable/1.3`) to the PR, in case that fails you need to create backports manually.

Testing:
* [ ] There are unit/integration tests that verify all acceptance criterias of the issue
* [ ] New tests are written to ensure backwards compatibility with further versions
* [x] The behavior is tested manually
* [ ] The change has been verified by a QA run
* [ ] The impact of the changes is verified by a benchmark

Documentation:
* [ ] The documentation is updated (e.g. BPMN reference, configuration, examples, get-started guides, etc.)
* [ ] New content is added to the [release announcement](https://drive.google.com/drive/u/0/folders/1DTIeswnEEq-NggJ25rm2BsDjcCQpDape)
* [ ] If the PR changes how BPMN processes are validated (e.g. support new BPMN element) then the Camunda modeling team should be informed to adjust the BPMN linting.

Please refer to our [review guidelines](https://github.com/camunda/zeebe/wiki/Pull-Requests-and-Code-Reviews#code-review-guidelines).


Co-authored-by: lzgabel <lz19960321lz@gmail.com>

dist/src/main/config/broker.standalone.yaml.template

@@ -585,17 +585,21 @@
         #     decision: true
         #     decisionEvaluation: true
         #     deployment: true
+        #     deploymentDistribution: true
         #     error: true
         #     incident: true
         #     job: true
         #     jobBatch: false
         #     message: true
+        #     messageStartSubscription: true
         #     messageSubscription: true
         #     process: true
+        #     processEvent: false
         #     processInstance: true
         #     processInstanceCreation: true
         #     processInstanceModification: true
         #     processMessageSubscription: true
+        #     timer: true
         #     variable: true
         #     variableDocument: true
 

dist/src/main/config/broker.yaml.template

@@ -523,17 +523,21 @@
         #     decision: true
         #     decisionEvaluation: true
         #     deployment: true
+        #     deploymentDistribution: true
         #     error: true
         #     incident: true
         #     job: true
         #     jobBatch: false
         #     message: true
+        #     messageStartSubscription: true
         #     messageSubscription: true
         #     process: true
+        #     processEvent: false
         #     processInstance: true
         #     processInstanceCreation: true
         #     processInstanceModification: true
         #     processMessageSubscription: true
+        #     timer: true
         #     variable: true
         #     variableDocument: true
 

exporters/elasticsearch-exporter/README.md

@@ -186,17 +186,21 @@ exporters:
         decision: true
         decisionEvaluation: true
         deployment: true
+        deploymentDistribution: true
         error: true
         incident: true
         job: true
         jobBatch: false
         message: true
+        messageStartSubscription: true
         messageSubscription: true
         process: true
+        processEvent: false
         processInstance: true
         processInstanceCreation: true
         processInstanceModification: true
         processMessageSubscription: true
+        timer: true
         variable: true
         variableDocument: true
 ```

exporters/elasticsearch-exporter/src/main/java/io/camunda/zeebe/exporter/ElasticsearchExporter.java

@@ -200,6 +200,18 @@ private void createIndexTemplates() {
       if (index.checkpoint) {
         createValueIndexTemplate(ValueType.CHECKPOINT);
       }
+      if (index.timer) {
+        createValueIndexTemplate(ValueType.TIMER);
+      }
+      if (index.messageStartEventSubscription) {
+        createValueIndexTemplate(ValueType.MESSAGE_START_EVENT_SUBSCRIPTION);
+      }
+      if (index.processEvent) {
+        createValueIndexTemplate(ValueType.PROCESS_EVENT);
+      }
+      if (index.deploymentDistribution) {
+        createValueIndexTemplate(ValueType.DEPLOYMENT_DISTRIBUTION);
+      }
     }
 
     indexTemplatesCreated = true;

exporters/elasticsearch-exporter/src/main/java/io/camunda/zeebe/exporter/ElasticsearchExporterConfiguration.java

@@ -89,6 +89,14 @@ public boolean shouldIndexValueType(final ValueType valueType) {
         return index.decisionEvaluation;
       case CHECKPOINT:
         return index.checkpoint;
+      case TIMER:
+        return index.timer;
+      case MESSAGE_START_EVENT_SUBSCRIPTION:
+        return index.messageStartEventSubscription;
+      case PROCESS_EVENT:
+        return index.processEvent;
+      case DEPLOYMENT_DISTRIBUTION:
+        return index.deploymentDistribution;
       default:
         return false;
     }
@@ -139,6 +147,10 @@ public static class IndexConfiguration {
     public boolean variableDocument = true;
 
     public boolean checkpoint = false;
+    public boolean timer = true;
+    public boolean messageStartEventSubscription = true;
+    public boolean processEvent = false;
+    public boolean deploymentDistribution = true;
 
     // index settings
     private Integer numberOfShards = null;
@@ -208,6 +220,14 @@ public String toString() {
           + decisionEvaluation
           + ", checkpoint="
           + checkpoint
+          + ", timer="
+          + timer
+          + ", messageStartEventSubscription="
+          + messageStartEventSubscription
+          + ", processEvent="
+          + processEvent
+          + ", deploymentDistribution="
+          + deploymentDistribution
           + '}';
     }
   }

exporters/elasticsearch-exporter/src/main/resources/zeebe-record-deployment-distribution-template.json

@@ -0,0 +1,32 @@
+{
+  "index_patterns": [
+    "zeebe-record_deployment-distribution_*"
+  ],
+  "composed_of": [
+    "zeebe-record"
+  ],
+  "priority": 20,
+  "version": 1,
+  "template": {
+    "settings": {
+      "number_of_shards": 1,
+      "number_of_replicas": 0,
+      "index.queries.cache.enabled": false
+    },
+    "aliases": {
+      "zeebe-record-deployment-distribution": {}
+    },
+    "mappings": {
+      "properties": {
+        "value": {
+          "dynamic": "strict",
+          "properties": {
+            "partitionId": {
+              "type": "integer"
+            }
+          }
+        }
+      }
+    }
+  }
+}

exporters/elasticsearch-exporter/src/main/resources/zeebe-record-message-start-event-subscription-template.json

@@ -0,0 +1,51 @@
+{
+  "index_patterns": [
+    "zeebe-record_message-start-event-subscription_*"
+  ],
+  "composed_of": ["zeebe-record"],
+  "priority": 20,
+  "version": 1,
+  "template": {
+    "settings": {
+      "number_of_shards": 1,
+      "number_of_replicas": 0,
+      "index.queries.cache.enabled": false
+    },
+    "aliases": {
+      "zeebe-record-message-start-event-subscription": {}
+    },
+    "mappings": {
+      "properties": {
+        "value": {
+          "dynamic": "strict",
+          "properties": {
+            "processDefinitionKey": {
+              "type": "long"
+            },
+            "processInstanceKey": {
+              "type": "long"
+            },
+            "messageName": {
+              "type": "keyword"
+            },
+            "correlationKey": {
+              "type": "text"
+            },
+            "bpmnProcessId": {
+              "type": "keyword"
+            },
+            "startEventId": {
+              "type": "keyword"
+            },
+            "messageKey": {
+              "type": "long"
+            },
+            "variables": {
+              "enabled": false
+            }
+          }
+        }
+      }
+    }
+  }
+}

exporters/elasticsearch-exporter/src/main/resources/zeebe-record-process-event-template.json

@@ -0,0 +1,42 @@
+{
+  "index_patterns": [
+    "zeebe-record_process-event_*"
+  ],
+  "composed_of": ["zeebe-record"],
+  "priority": 20,
+  "version": 1,
+  "template": {
+    "settings": {
+      "number_of_shards": 1,
+      "number_of_replicas": 0,
+      "index.queries.cache.enabled": false
+    },
+    "aliases": {
+      "zeebe-record-process-event": {}
+    },
+    "mappings": {
+      "properties": {
+        "value": {
+          "dynamic": "strict",
+          "properties": {
+            "scopeKey": {
+              "type": "long"
+            },
+            "processInstanceKey": {
+              "type": "long"
+            },
+            "processDefinitionKey": {
+              "type": "long"
+            },
+            "targetElementId": {
+              "type": "keyword"
+            },
+            "variables": {
+              "enabled": false
+            }
+          }
+        }
+      }
+    }
+  }
+}

exporters/elasticsearch-exporter/src/main/resources/zeebe-record-timer-template.json

@@ -0,0 +1,45 @@
+{
+  "index_patterns": [
+    "zeebe-record_timer_*"
+  ],
+  "composed_of": ["zeebe-record"],
+  "priority": 20,
+  "version": 1,
+  "template": {
+    "settings": {
+      "number_of_shards": 1,
+      "number_of_replicas": 0,
+      "index.queries.cache.enabled": false
+    },
+    "aliases": {
+      "zeebe-record-timer": {}
+    },
+    "mappings": {
+      "properties": {
+        "value": {
+          "dynamic": "strict",
+          "properties": {
+            "elementInstanceKey": {
+              "type": "long"
+            },
+            "dueDate": {
+              "type": "long"
+            },
+            "targetElementId": {
+              "type": "keyword"
+            },
+            "repetitions": {
+              "type": "integer"
+            },
+            "processDefinitionKey": {
+              "type": "long"
+            },
+            "processInstanceKey": {
+              "type": "long"
+            }
+          }
+        }
+      }
+    }
+  }
+}

exporters/elasticsearch-exporter/src/test/java/io/camunda/zeebe/exporter/TestSupport.java

@@ -67,6 +67,10 @@ static void setIndexingForValueType(
       case DECISION_REQUIREMENTS -> config.decisionRequirements = value;
       case DECISION_EVALUATION -> config.decisionEvaluation = value;
       case CHECKPOINT -> config.checkpoint = value;
+      case TIMER -> config.timer = value;
+      case MESSAGE_START_EVENT_SUBSCRIPTION -> config.messageStartEventSubscription = value;
+      case PROCESS_EVENT -> config.processEvent = value;
+      case DEPLOYMENT_DISTRIBUTION -> config.deploymentDistribution = value;
       default -> throw new IllegalArgumentException(
           "No known indexing configuration option for value type " + valueType);
     }
@@ -97,14 +101,7 @@ static void setIndexingForRecordType(
    */
   static Stream<ValueType> provideValueTypes() {
     final var excludedValueTypes =
-        EnumSet.of(
-            ValueType.SBE_UNKNOWN,
-            ValueType.NULL_VAL,
-            ValueType.TIMER,
-            ValueType.PROCESS_INSTANCE_RESULT,
-            ValueType.DEPLOYMENT_DISTRIBUTION,
-            ValueType.PROCESS_EVENT,
-            ValueType.MESSAGE_START_EVENT_SUBSCRIPTION);
+        EnumSet.of(ValueType.SBE_UNKNOWN, ValueType.NULL_VAL, ValueType.PROCESS_INSTANCE_RESULT);
     return EnumSet.complementOf(excludedValueTypes).stream();
   }
 }