Configuring Elasticsearch indexing

Before proceeding, it is recommended to familiarize yourself with Elasticsearch and its indexing process by referring to the Intro to Elasticsearch section.

Intro to Elasticsearch

Configuration

Old configuration:

spring:
  elasticsearch:
    index-settings:
      name: process_instance
      shards: 2
      replicas: 2

New configuration:

flowx:
  indexing:
    optimistic-locking-retries: 3 //controls how many times the engine will retry indexing when it encounters concurrent-update errors on key identifiers.  
    enabled: true  // true | false - specifies if the indexing with Elasticsearch for the whole app is enabled or disabled.
    processInstance:  // set of configurations for indexing process instances. Can be duplicated for other objects.
      indexing-type: kafka  // no-indexing | http | kafka - the chosen indexing strategy.
      index-name: process_instance  // the index name that is part of the search pattern.
      shards: 1
      replicas: 1

The flowx.indexing.optimistic-locking-retries property is available only for the HTTP indexing strategy.

The flowx.indexing.enabled property determines whether indexing with Elasticsearch is enabled. When set to false or missing, no indexing will be performed for any entities defined below. When set to true, indexing with Elasticsearch is enabled.

If the FlowX indexing configuration is set to false, the following configuration information and guidelines are not applicable to your use case.

The flowx.indexing.processInstance.indexing-type property defines the indexing strategy for process instances. It can have one of the following values:

no-indexing: No indexing will be performed for process instances.
http: Direct connection from the process engine to Elasticsearch through HTTP calls.
kafka: Data will be sent to be indexed via a Kafka topic using the new strategy. To implement this strategy, the Kafka Connect with Elasticsearch Sink Connector must be deployed in the infrastructure.

Configuration steps

To enable indexing with Elasticsearch for the entire application, update the process-engine configuration with the following parameters:

FLOWX_INDEXING_ENABLED: Set this parameter to true to enable indexing with Elasticsearch for the entire application.

Variable Name	Enabled	Description
FLOWX_INDEXING_ENABLED	true	Indexing with Elasticsearch for the whole app is enabled
FLOWX_INDEXING_ENABLED	false	Indexing with Elasticsearch for the whole app is disabled

FLOWX_INDEXING_PROCESSINSTANCE_INDEXING_TYPE: Set this parameter to kafka to use the Kafka transport strategy for indexing process instances.

Variable Name	Indexing Type - Values	Definition
FLOWX_INDEXING_PROCESSINSTANCE_INDEXING_TYPE	no-indexing	No indexing is performed for process instances
FLOWX_INDEXING_PROCESSINSTANCE_INDEXING_TYPE	http	Process instances are indexed via HTTP (direct connection from process-engine to Elasticsearch thorugh HTTP calls)
FLOWX_INDEXING_PROCESSINSTANCE_INDEXING_TYPE	kafka	Process instances are indexed via Kafka (send data to be indexed through a kafka topic - the new strategy for the applied solution)

FLOWX_INDEXING_PROCESSINSTANCE_INDEX_NAME: Specify the name of the index used for process instances.

Variable Name	Values	Definition
FLOWX_INDEXING_PROCESSINSTANCE_INDEXING_INDEX_NAME	process_instance	The name of the index used for storing process instances. It is also part of the search pattern

FLOWX_INDEXING_PROCESSINSTANCE_SHARDS: Set the number of shards for the index.

Variable Name	Values	Definition
FLOWX_INDEXING_PROCESSINSTANCE_SHARDS	1	The number of shards for the Elasticsearch index storing process instances

FLOWX_INDEXING_PROCESSINSTANCE_REPLICAS: Set the number of replicas for the index.

Variable Name	Values	Definition
FLOWX_INDEXING_PROCESSINSTANCE_REPLICAS	1	The number of replicas for the Elasticsearch index storing process instances

For Kafka indexing, the Kafka Connect with Elasticsearch Sink Connector must be deployed in the infrastructure.

Elasticsearch Service Sink Connector

Configuration examples

Kafka Connect

Assumes Kafka cluster installed with Strimzi operator and Elasticsearch with eck-operator
Can save the image built by Kafka Connect to a local registry and comment build section

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: kafka-connect-kafka-flowx
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 3.0.0
  replicas: 1
  bootstrapServers: kafka-flowx-kafka-bootstrap:9093
  tls:
    trustedCertificates:
      - secretName: kafka-flowx-cluster-ca-cert
        certificate: ca.crt
  image: ttl.sh/strimzi-connect-ttlsh266-3.0.0:24h
  config:
    group.id: flowx-kafka-connect
    offset.storage.topic: kafka-connect-cluster-offsets
    config.storage.topic: kafka-connect-cluster-configs
    status.storage.topic: kafka-connect-cluster-status
    # -1 means it will use the default replication factor configured in the broker
    config.storage.replication.factor: -1
    offset.storage.replication.factor: -1
    status.storage.replication.factor: -1
    topic.creation.enable: true
  build:
   output:
     type: docker
     # This image will last only for 24 hours and might be overwritten by other users
     # Strimzi will use this tag to push the image. But it will use the digest to pull
     # the container image to make sure it pulls exactly the image we just built. So
     # it should not happen that you pull someone else's container image. However, we
     # recommend changing this to your own container registry or using a different
     # image name for any other than demo purposes.
     image: ttl.sh/strimzi-connect-ttlsh266-3.0.0:24h
   plugins:
     - name: kafka-connect-elasticsearch
       artifacts:
         - type: zip
           url: https://d1i4a15mxbxib1.cloudfront.net/api/plugins/confluentinc/kafka-connect-elasticsearch/versions/14.0.6/confluentinc-kafka-connect-elasticsearch-14.0.6.zip
  externalConfiguration:
    volumes:
      - name: elasticsearch-keystore-volume
        secret:
          secretName: elasticsearch-keystore
    env:
      - name: SPRING_ELASTICSEARCH_REST_PASSWORD
        valueFrom:
          secretKeyRef:
            name: elasticsearch-es-elastic-user
            key: elastic

Kafka Elasticsearch connector

spec:
  class: io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
  tasksMax: 2
  config:
    tasks.max: "2" # The maximum number of tasks that can be run in parallel for this connector, which is 2 in this case. You can start with 2 and increase in case of huge load, but pay attention to the load that will be produced on Elasticsearch and also to the resources that you allocate to KafkaConnect so that it can support the threads.
    topics: "ai.flowx.core.index.process.v1" # Source Kafka topic. Must be the same as the one declared in the process defined as ${kafka.topic.naming.prefix}.core.index.process${kafka.topic.naming.suffix}
    key.ignore: "false" # Don't change this value! This tells Kafka Connect (KC) to process the key of the message - it will be used as the ID of the object in Elasticsearch.  
    schema.ignore: "true" # Don't change this value!!! This tells KC to ignore the mapping from the Kafka message. Elasticsearch will use internal mapping. See below. 
    connection.url: "https://elasticsearch-es-http:9200" # The URL to Elasticsearch. You should configure this.
    connection.username: "elastic" # The username to authenticate with Elasticsearch. You should configure this.
    connection.password: "Yyh03ZI66310Hyw59MXcR8xt" # The password to authenticate with Elasticsearch. You should configure this.
    elastic.security.protocol: "SSL" # The security protocol to use for connecting to Elasticsearch. You should use SSL if possible.
    elastic.https.ssl.keystore.location: "/opt/kafka/external-configuration/elasticsearch-keystore-volume/keystore.jks" # You should configure the path to the keystore where the Elasticsearch key is added.
    elastic.https.ssl.keystore.password: "MPx57vkACsRWKVap" # The password for the keystore file. You should configure this.
    elastic.https.ssl.key.password: "MPx57vkACsRWKVap" # The password for the key within the keystore file.
    elastic.https.ssl.keystore.type: "JKS" # The type of the keystore file. It is set to "JKS" (Java KeyStore).
    elastic.https.ssl.truststore.location: "/opt/kafka/external-configuration/elasticsearch-keystore-volume/keystore.jks" # you should configure the path to the keystore where the Elasticsearch key is added.
    elastic.https.ssl.truststore.password: "MPx57vkACsRWKVap" # The password for the truststore file. You should configure this
    elastic.https.ssl.truststore.type: "JKS" # The type of the truststore file. It is set to "JKS".
    elastic.https.ssl.protocol: "TLS" # The SSL/TLS protocol to use for communication. It is set to "TLS".
    batch.size: 1000 # The size of the message batch that KC will process. This should be fine as default value. If you experience slowness and want to increase the speed, changing this may help but it will be based on your scenario. Consult the documentation for more details.
    linger.ms: 1 # #Start with this value and change it only if needed. Consult the documentation for connector for more details.
    read.timeout.ms: 30000 # Increased to 30000 from the default 3000 due to flush.synchronously = true.
    flush.synchronously: "true" # Don't change this value! The way of writing to Elasticsearch. It must stay "true" for the router below to work.
    drop.invalid.message: "true" # Don't change this value! If set to false, the connector will wait for a configuration that allows processing the message. If set to true, the connector will drop the invalid message.
    behavior.on.null.values: "IGNORE" # Don't change this value! Must be set to IGNORE to avoid blocking the processing of null messages.
    behavior.on.malformed.documents: "IGNORE" # Don't change this value!  Must be IGNORE to avoid blocking the processing of invalid JSONs.
    write.method: "UPSERT" # Don't change this value!  UPSERT to create or update the index.
    type.name: "_doc" # Don't change this value! This is the name of the Elasticsearch type for indexing. 
    key.converter: "org.apache.kafka.connect.storage.StringConverter" # Don't change this value!
    key.converter.schemas.enable: "false" # Don't change this value! No schema defined for the key in the message.
    value.converter: "org.apache.kafka.connect.json.JsonConverter" # Don't change this value!
    value.converter.schemas.enable: "false" # Don't change this value! No schema defined for the value in the message body.
    transforms: "routeTS" # Don't change this value! This represents router that helps create indices dynamically based on the timestamp (process instance start date).
    transforms.routeTS.type: "org.apache.kafka.connect.transforms.TimestampRouter" # Don't change this value! It helps with routing the message to the correct index.
    transforms.routeTS.topic.format: "process_instance-${timestamp}" # You should configure this. It is important that this value must start with the value defined in process-engine config: flowx.indexing.processInstance.index-name. The name of the index will start with a prefix ("process_instance-" in this example) and must have the timestamp appended after for dynamically creating indices. For backward compatibility (utilizing the data in the existing index), the prefix must be "process_instance-". However, backward compatibility isn't specifically required here.
    transforms.routeTS.timestamp.format: "yyyyMMdd" # This format ensures that the timestamp is represented consistently and can be easily parsed when creating or searching for indices based on the process instance start date. You can change this with the value you want. If you want monthly indexes set it to "yyyyMM". But be aware that once you set it, when you change it, the existing object indexed will not be updated anymore. The update messages will be treated as new objects and indexed again because they are going to be sent to new indexes. This is important! Try to find your index size and stick with it.

HTTP indexing

flowx:
  indexing:
    enabled: true
    processInstance:
      indexing-type: http
      index-name: process_instance
      shards: 2
      replicas: 2

If you don’t want to remove the existing configuration parameters, you can use the following example:

spring:
  elasticsearch:
    index-settings:
      name: process_instance
      shards: 2
      replicas: 2
flowx.use-elasticsearch: true
flowx:
  indexing:
    enabled: ${flowx.use-elasticsearch}
    processInstance:
      indexing-type: http
      index-name: ${spring.elasticsearch.index-settings.name}
      shards: ${spring.elasticsearch.index-settings.shards}
      replicas: ${spring.elasticsearch.index-settings.replicas}

Querying Elasticsearch

To read from multiple indices, queries in Elasticsearch have been updated. The queries now run against an index pattern that identifies multiple indices instead of a single index. The index pattern is derived from the value defined in the configuration property: flowx.indexing.processInstance.index-name

Kafka topics - process events messages

This topic is used for sending the data to be indexed from Process engine. The data from this topic will be read by Kafka Connect.

Key: ${kafka.topic.process.index.out}
Value: ${kafka.topic.naming.prefix}.core.index.process${kafka.topic.naming.suffix}

Default parameter (env var)	Default FLOWX.AI value (can be overwritten)
KAFKA_TOPIC_PROCESS_INDEX_OUT	ai.flowx.dev.core.index.process.v1

The topic name, defined in the value, will be used by Kafka Connect as source for the messages to be sent to Elasticsearch for indexing.The attribute indexLastUpdatedTime is new and will be populated for the kafka-connect strategy. This will tell the timestamp when the last operation was done on the object in the index.

Elasticsearch index template

The mappings between messages and Elasticsearch data types need to be specified through index templates. The process engine automatically handles template creation during startup, but the approach differs based on the indexing strategy:

When using the HTTP indexing strategy (indexing-type: http), the process engine:
- Automatically creates the index with the template applied
- Applies the number of shards and replicas directly from the configuration
When using the Kafka indexing strategy (indexing-type: kafka), the process engine:
- Creates an index template (if it doesn’t exist) that Elasticsearch will use for dynamic index creation
- The template applies to indices matching the pattern derived from flowx.indexing.processInstance.index-name
- Kafka Connect creates indices dynamically based on the timestamp, and Elasticsearch applies the template automatically
- The number of shards and replicas are set dynamically based on configuration parameters

The template creation is fully automated by the process engine. No manual intervention is required for template management. The number_of_shards and number_of_replicas values are automatically populated from your environment configuration (FLOWX_INDEXING_PROCESSINSTANCE_SHARDS and FLOWX_INDEXING_PROCESSINSTANCE_REPLICAS).

For the Kafka indexing strategy, the indexLastUpdatedTime attribute is automatically populated to track when the last operation was performed on the object in the index.

Time-based partitioning and index deletion

When working with large volumes of data, it’s recommended to implement time-based partitioning for Elasticsearch indices to improve performance and manageability.

Partitioning with Kafka vs HTTP

While both HTTP and Kafka indexing strategies support basic Elasticsearch sharding, only the Kafka strategy provides out-of-the-box support for time-based partitioning through the transforms.routeTS.timestamp.format in the Kafka Sink Connector configuration.

Time-based partitioning (creating separate indices for different time periods like daily/weekly/monthly) is not available as a built-in feature when using the HTTP indexing strategy. For efficient time-based partitioning and index lifecycle management, we recommend using the Kafka indexing strategy.

Efficient data deletion

When deleting data from Elasticsearch, it’s significantly more efficient to delete entire indices rather than deleting individual documents. This is particularly important for maintaining performance in systems with high data volumes. The Kafka indexing strategy automatically creates time-based indices that can be deleted as entire units when they’re no longer needed. This aligns well with database partitioning strategies, allowing for consistent data lifecycle management across your database and Elasticsearch. For optimal performance, align your Elasticsearch time-based partitioning with your database partitioning strategy:

Database partitioning	Recommended Elasticsearch timestamp format
`MONTH`	`yyyyMM` (monthly indices)
`WEEK`	`yyyyww` (weekly indices)
`DAY`	`yyyyMMdd` (daily indices)

Here are some guidelines to help you get started:

Microservices

Plugins

Observability

Microservices access rights

Plugins access rights

Configuring Elasticsearch indexing

Intro to Elasticsearch

Configuration

Configuration steps

Configuration examples

Kafka Connect

Kafka Elasticsearch connector

HTTP indexing

Querying Elasticsearch

Kafka topics - process events messages

Elasticsearch index template

Time-based partitioning and index deletion

Partitioning with Kafka vs HTTP

Efficient data deletion

Configuration guidelines

Intro to Elasticsearch

​Configuration

​Configuration steps

​Configuration examples

​Kafka Connect

​Kafka Elasticsearch connector

​HTTP indexing

​Querying Elasticsearch

​Kafka topics - process events messages

​Elasticsearch index template

​Time-based partitioning and index deletion

​Partitioning with Kafka vs HTTP

​Efficient data deletion

Configuration guidelines

Configuration

Configuration steps

Configuration examples

Kafka Connect

Kafka Elasticsearch connector

HTTP indexing

Querying Elasticsearch

Kafka topics - process events messages

Elasticsearch index template

Time-based partitioning and index deletion

Partitioning with Kafka vs HTTP

Efficient data deletion