Ingesting data from Kafka into Mach5

This document explains how to ingest Kafka records into a Mach5 Search index by configuring a Kafka connection and ingest pipeline in the Mach5 UI. Additional formats are documented in release-specific sections below.

Prerequisites

• This document assumes that Mach5 is deployed and running successfully. Let’s assume the Mach5 UI is running at http://localhost:8888/.

• Store, store route and warehouse are created successfully. Refer to Quickstart document for help

• Consider there is a Kafka topic named kafkatopic already set up having following 10 records:

@kafka-client$ kafka-console-consumer.sh --bootstrap-server kafka.kafka.svc.cluster.local:9092 --topic kafkatopic --from-beginning
{ "id": "1", "ip_address": "192.168.1.1" } 
{ "id": "2", "ip_address": ["192.168.1.2", "10.0.0.5"] } 
{ "id": "3", "ip_address": "10.0.0.1" }
{ "id": "4", "ip_address": ["10.0.0.2", "8.8.8.8"] } 
{ "id": "5", "ip_address": "172.16.0.1" }
{ "id": "6", "ip_address": ["172.16.0.2", "192.168.1.1"] } 
{ "id": "7", "ip_address": "8.8.8.8" }
{ "id": "8", "ip_address": ["8.8.4.4", "10.0.0.5"] }
{ "id": "9", "ip_address": "192.168.2.1" } 
{ "id": "10", "ip_address": ["192.168.2.2", "8.8.4.4"] }
Processed a total of 10 messages

• Mach5 Index is already created with the name kafkaindex and with relevant mappings for the kafka topic. If no mappings are provided, Mach5 will dynamically infer the mapping when ingesting data. Please note that some data may not get ingested if mappings are not provided.

Connections

A connection stores the properties needed to access a source such as Kafka. The same Kafka connection can be used by multiple ingest pipelines.

Click on Connections on the left panel of Mach5 UI

Add a new connection

Click Create Connection, select Kafka, and fill in the connection details.

• Name: Provide a name for the connection, for example kafka-conn.
• Bootstrap Servers: Specify the Kafka bootstrap server, for example kafka.kafka.svc.cluster.local:9092.
• Authentication: Select None, SASL, or AWS_MSK_IAM based on your Kafka deployment.
• Enable Protocol SSL: Select this only when the Kafka connection requires SSL.
• Enable Verify Certificate: Select this only when Mach5 should verify the Kafka broker certificate.
• Test: Click Test to verify connectivity to Kafka.
• Click on Save

Ingest Pipelines

Mach5 Search ingest pipelines process and ingest documents from sources such as Kafka, S3, and Iceberg. A Kafka ingest pipeline uses a Kafka connection to read records from a topic and write decoded documents into a Mach5 index.

Click on Ingest Pipelines on the left panel of Mach5 UI

Add an ingest pipeline

Click Create Pipeline on the Ingest Pipelines page.

• Name: Provide a name for the ingest pipeline, for example kafka-ip.
• Index: Specify the target Mach5 index, for example kafkaindex. The index must exist before enabling the ingest pipeline.
• Enabled: Select this checkbox if ingestion should start after the pipeline is saved.
• Transform Type: Select None, JavaScript, or VRL. Use a transform when records need to be changed before indexing.
• Connection Name: Select the Kafka connection name that was created earlier

• Ingest Pipeline Type: Select Kafka.
• Topic: Specify the Kafka topic to ingest from, for example kafkatopic.
• Data Format: Select JSON for JSON record values.
• Start Offset: Select Earliest to read from the beginning of the topic, or Latest to read only new records.
• Group Id: Specify the Kafka consumer group ID, for example gidkafka.
• Compression: Select None, gzip, or zstd based on the Kafka record value compression.
• Offsets Per Batch: Default is 16384.
• Chunk Size: Default is 16384.
• Advanced: Per-Partition Offsets: Use this only when specific partitions need explicit starting offsets.
• Advanced: Leave advanced pipeline settings as default unless you need to change settings such as Operation Mode or Ignore Mapping Errors.
• Click Save.
• Once the ingest pipeline is created, records will start getting reflected in the Mach5 index corresponding to the Kafka topic.

Verify an ingest pipeline

After saving, verify that the pipeline appears on the Ingest Pipelines page.

Once the kafka-ip ingestion pipeline is successfully created, records from Kafka in kafkatopic are ingested into Mach5 index kafkaindex. As and when new data comes to the Kafka topic, it will get added into the Mach5 index

Verify data ingestion

Use a Mach5 notebook with an OpenSearch DSL cell to verify that records from Kafka are ingested into Mach5.

• Open Notebooks from the left panel.
• Click Create Notebook and provide a notebook name, for example kafka-verify.
• Open the notebook and select the warehouse to use for querying.
• Click Add Cell and select Add OpenSearch DSL Cell.

Run a count query in the OpenSearch DSL cell to verify the number of ingested records:

GET /kafkaindex/_count
{
  "query": {
    "match_all": {}
  }
}

For the sample data in Prerequisites, the expected count is 10.

To inspect the ingested records, add another OpenSearch DSL cell or replace the count query with this search query:

GET /kafkaindex/_search
{
  "size": 10,
  "query": {
    "match_all": {}
  }
}

For the sample data in Prerequisites, the search response should show the 10 ingested records from kafkatopic.

Disable ingest pipeline

When not in use, an ingest pipeline can be disabled so updates to source data are not reflected in Mach5.

• To Disable an existing pipeline, click on the Edit icon next to the the ingest pipeline, eg. kafka-ip
• In the Edit ingest pipeline page, at end of all options, before Save button, deselect the Enabled checkbox
• Save the ingest pipeline to take effect
• The kafka-ip ingest pipeline is now disabled. It will not read data from source to ingest data into Mach5. It can be enabled any time it needs to be re-used

Kafka Tombstone Support in Mach5

Tombstone Messages

In Kafka, tombstone messages are messages with a valid, non-empty key and an empty/null payload (no value). Tombstone messages can be added to a topic by any producer and are often used in conjunction with compacted topics in order to delete previous instances of a particular key.

Transform functions are optional in Mach5 ingest pipelines. However, when ingesting Kafka tombstone messages, a transform function is required to explicitly define the desired tombstone handling behavior. When Mach5 detects a tombstone message, the meta.kafka.op field is set to "delete" for the record in the ingest pipeline. This field can be interpreted by the transform function in order to produce one of two outcomes:

1. Delete: When the transform function returns a record containing a single _id field, the corresponding document in the index with the matching _id field (if it exists) will be deleted. Including any fields other than _id will result in an error.
2. Ignore: When the transform function returns the empty array [], no action will be performed on the index.

Examples

Deleting with Tombstone Messages

The following transform function shows how to (explicitly) interpret and act upon a delete operation.

Note: The following example assumes that the Kafka record key is non-null and UTF-8–encoded, and that it maps directly to the document _id. This is a common pattern, but not a requirement. Applications with different key encodings or error-handling needs should decode and validate the key accordingly (for example, using try/catch).

function transform(payload, meta) {
  const kafka = meta?.kafka;
  
  // Assumes the key is not null and is a valid UTF8 string
  const id = decodeUtf8(new Uint8Array(kafka.key));
  
  if (kafka?.op === "delete") {
    return { _id: id };
  } else {
    // Further processing and transformation of payload
    return { ...payload, _id: id};
  }
}

const rec = transform(arg, meta);
rec

A more compact version of this function can be written to take advantage of JavaScript object spreading semantics:

function transform(payload, meta) {
  const kafka = meta?.kafka;
  const id = decodeUtf8(new Uint8Array(kafka.key));
  return { ...payload, _id: id };
}

const rec = transform(arg, meta);
rec

When the operation is a delete, arg/payload is null, so spreading is a no-op in JavaScript, producing a single object with the _id field as required.

Ignoring Tombstone Messages

The following example shows how to explicitly ignore tombstone messages while continuing to process non-tombstone records:

function transform(payload, meta) {
  const kafka = meta?.kafka;

  if (kafka?.op === "delete") {
    return [];
  }

  // Normal processing for non-tombstone records
  return payload;
}

Kafka Protobuf Support in Mach5

Mach5 Kafka ingest pipelines can decode Protobuf-encoded Kafka record values when the pipeline is configured with a compiled descriptor set and a message type.

Why Mach5 Needs a Descriptor Set

Mach5 receives the Kafka record value as bytes. For Protobuf records, Mach5 needs the compiled Protobuf FileDescriptorSet so it can resolve field names, field types, nested messages, and the selected top-level message type during ingestion.

Mach5 expects:

• Descriptor Set: A compiled FileDescriptorSet, not a raw .proto file.
• Message Type: The fully qualified top-level Protobuf message type to decode from each Kafka record value, for example it.kafka.Event.
• Kafka Value: The Kafka record value must match the selected message type and descriptor set.

Compile the descriptor set before configuring the Mach5 ingest pipeline:

protoc -I . --descriptor_set_out=event.desc --include_imports event.proto

Pipeline Configuration

For a Kafka ingest pipeline that reads Protobuf values, set Data Format to Protobuf and provide:

• Protobuf Descriptor: Upload the compiled descriptor set.
• Message Type: Select the top-level message type from the uploaded descriptor.
• Unwrap Targets: Configure only when a bytes field contains another serialized Protobuf message that Mach5 should decode.

The equivalent API fields are:

"data_format": {
  "type": "protobuf",
  "file_descriptor_set": "<base64-encoded descriptor set>",
  "message_type": "it.kafka.Event"
}

Unwrap Targets

Use unwrap_targets when the selected top-level Protobuf message contains bytes fields that should be decoded as nested Protobuf messages. Mach5 preserves the surrounding decoded document and replaces the configured bytes fields with decoded objects.

Example:

"data_format": {
  "type": "protobuf",
  "file_descriptor_set": "<base64-encoded descriptor set>",
  "message_type": "it.kafka.Outer",
  "unwrap_targets": [
    { "path": ["payload"], "message_type": "it.kafka.Middle" },
    { "path": ["payload", "body"], "message_type": "it.kafka.Event" }
  ]
}

Each unwrap target requires:

• Path: The field path to the bytes field in the decoded document.
• Message Type: The fully qualified Protobuf message type to use for decoding that bytes field.

More Options

The standard Kafka ingest pipeline options still apply to Protobuf pipelines, including Start Offset, Group Id, Offsets Per Batch, Chunk Size, Operation Mode, Transform Type, and Enabled.

Transform functions run after Mach5 decodes the Protobuf value. Use a transform function if you need to set _id, rename decoded fields, drop records, or apply tombstone handling.

Troubleshooting

Issue: Ingest pipeline does not work and no new pods are being spawned

• Symptom: The ingest pipeline appears to be configured but no data is ingested.
• Cause: The Mach5 index referenced in the ingest pipeline does not exist.
• Resolution: Ensure that the required Mach5 index is created before configuring or enabling the ingest pipeline.

Issue: Protobuf pipeline is enabled but no documents are indexed

• Symptom: Kafka has records, but Mach5 does not show decoded documents.
• Cause: The descriptor set, selected message type, or Kafka record value does not match.
• Resolution: Verify that the descriptor set was compiled from the schema used by the producer, and that Message Type exactly matches the fully qualified top-level Protobuf message type.

Issue: Nested Protobuf fields are still bytes

• Symptom: A decoded Protobuf document still contains bytes for a nested field.
• Cause: The nested bytes field is not configured as an unwrap target, or the unwrap target path does not match the decoded field path.
• Resolution: Add or correct the unwrap target with the field path and nested message type.

Issue: An ingest pipeline run failed

• Symptom: The pipeline’s Runs page shows a Failed run.
• Cause: Open the run. The failure shows a cause badge (for example Evicted, Worker OOM, Kubernetes API or Terminated) and the worker, partition and offset range involved. The Workers list shows how each worker ended.
• Resolution: For more detail, or for runs that are no longer listed, query the _system/mach5-ingestion-audit-v1 index as described in Observability in Mach5. On EKS, repeated Evicted or Terminated causes usually come from node rebalancing or node drains; see “Avoid node terminations that interrupt ingestion” in EKS Cluster configuration.