---
title: Ingesting Kafka data in a Kubernetes installation
sidebar_label: Kafka in Kubernetes
---

This guide demonstrates how to set up an in-worker [Data Import Server (DIS)](../../data-guide/dis.md) that connects to a Kafka broker and writes incoming data to a Deephaven table for a system deployed on Kubernetes.

### Benefits of Kafka with Deephaven on Kubernetes

- **Scalability**: Leverage Kubernetes orchestration to scale Kafka ingestion based on workload demands
- **Resilience**: Benefit from Kubernetes self-healing capabilities for more reliable data ingestion
- **Portability**: Deploy your Kafka ingestion pipelines consistently across different Kubernetes environments
- **Resource Optimization**: Configure resource limits and requests for optimized performance

### Core+ vs. Legacy Kafka Ingestion

Deephaven provides two frameworks for Kafka ingestion:

- **Core+ Kafka Ingestion**: The newer, recommended framework with improved performance and features
- **Legacy Kafka Ingestion**: The traditional Enterprise Kafka framework

This guide supports both frameworks, with specific notes about Core+ benefits where relevant. For comprehensive details on Core+ Kafka ingestion, see [Core+ Kafka documentation](../../data-guide/streaming/coreplus-kafka.md).

### Prerequisites

- Access to a Kubernetes cluster with Deephaven deployed
- Familiarity with Kubernetes and the `kubectl` command
- Existing Kafka topic(s) accessible from your Kubernetes cluster

The examples in this guide use a namespace of `deephaven` - adjust accordingly for your environment.

## Architecture

In this architecture:

1. A Persistent Query runs inside a worker pod in the Kubernetes cluster
2. The Persistent Query initializes a Data Import Server (DIS) process
3. The DIS connects to external Kafka broker(s) and consumes topic data
4. Data is stored in a Persistent Volume Claim (PVC) for durability
5. The Table Data Cache Proxy (TDCP) routes queries to the appropriate data sources

## Configure a new Data Import Server

### Open a management shell terminal

Open a terminal session on management-shell pod.

```shell
$ kubectl -n deephaven exec -it deploy/management-shell -- /bin/bash
```

### Create a new DIS configuration

Next, use the [dhconfig dis](../../sys-admin/configuration/dhconfig/dis.md) tool to add a new data import
server named `iwd-kafka1` with a [claim](../../sys-admin/configuration/yaml.md#claims) for all tables in the `Kafka` namespace
with the below command. This configures Deephaven's [data routing](../../sys-admin/configuration/yaml.md) without requiring
any manual config file changes. The `--name` parameter value is arbitrary, but make a note of it. It will be needed when we
create the persistent query script in a Deephaven console later.

```shell
root@management-shell-1a2b3c4d5e-vwxyz:/$ /usr/illumon/latest/bin/dhconfig dis add --name iwd-kafka1 --claim Kafka
```

### Restart the Table Data Cache Proxy (TDCP)

After the routing file changes, the `tdcp` process needs a restart to pick up the changes. Note that this is _not_ done within
a management-shell terminal. We will perform a scale down and scale up of the `tdcp` deployment to do this with these commands.

```shell
# Scale the tdcp deployment down and wait a moment for the pod to terminate:
$ kubectl -n deephaven scale deploy/tdcp --replicas 0

# Scale the tdcp deployment back up
$ kubectl -n deephaven scale deploy/tdcp --replicas 1
```

## Create the in-worker DIS as a Persistent Query

### Create a schema

If you do not have a table to consume your Kafka data, you will need to create a schema for the Deephaven table. The recommended approach is to use the `SchemaHelper` tool, which automatically generates a schema from your `KafkaTableWriter.Options` configuration.

In your Persistent Query script, after configuring your options with key and value specifications, add the schema helper call before `consumeToDis`:

```groovy
import io.deephaven.enterprise.kafkawriter.SchemaHelper

// After configuring opts with keySpec and valueSpec...

// Automatically create or validate the schema based on the Options
new SchemaHelper(opts).addOrValidateSchema()

// Then start ingestion
KafkaTableWriter.consumeToDis(opts)
```

The `SchemaHelper` derives the schema from your `keySpec` and `valueSpec` configurations. You can customize the generated schema with grouping columns, symbol table settings, and merge key formulas:

```groovy
new SchemaHelper(opts)
    .withGroupingCols("Symbol", "Exch")
    .withSymbolTableNone("Flags")
    .withMergeKeyFormula("${autobalance_by_first_grouping_column}")
    .addOrValidateSchema()
```

For more details about schema creation options, see [schema helper tools](./coreplus-kafka.md#schema-helper-tools) in the Core+ Kafka documentation.

### Create a Persistent Query

Use the Deephaven web console and create a [Persistent Query](../../interfaces/web/query-monitor.md).

#### Enter the settings

Under the Settings tab, click `Show Advanced` and fill in the Persistent Volume Claim, Storage Class, Storage Size, and
Mount Path fields. If the stated persistent volume claim (pvc) does not exist, one is created with the storage class and
size specified. In that case, the storage class you specify must be one that has a dynamic volume provisioner associated with
it so a persistent volume is also created. If you choose to use a pre-existing pvc, you do not need to specify
the storage class or storage size.

![img](../../assets/kubernetes/k8s-kafka-pq-web.png)

#### Write the script

Click the Script tab and enter a script to create the data import server that connects to your Kafka broker. A simple
example script is shown below, and there are further examples and more detailed information on consuming Kafka data
for [Core+ workers](./coreplus-kafka.md) and [Legacy workers](../../legacy/importing-data/kafka.md#create-an-import-script).

```groovy
import io.deephaven.kafka.KafkaTools
import io.deephaven.enterprise.kafkawriter.KafkaTableWriter
import io.deephaven.enterprise.dataimportserver.DataImportServerTools

kafkaServer="kafka-broker-example.com"
kafkaTopicName="quickstart"         // Your Kafka topic
kafkaDisName="iwd-kafka1"           // The storage 'name' value you added to the routing.yml file
targetNamespace="Kafka"             // The target namespace added to the 'claims' section of the routing.yml file
targetTable="IWD_Test"              // The target table for which you created a schema

// Set Kafka properties
final Properties props = new Properties()

// Connection settings
props.put("bootstrap.servers", kafkaServer + ":9092")
props.put("group.id", "dhdis-k8s")

// Performance tuning
props.put("fetch.min.bytes", "65000")          // Minimum amount of data to fetch in a single request
props.put("fetch.max.wait.ms", "200")         // Maximum time to wait before returning data
props.put("fetch.max.bytes", "52428800")      // Maximum bytes to fetch per partition (50MB)
props.put("max.partition.fetch.bytes", "1048576") // Maximum bytes per partition (1MB)
props.put("max.poll.records", "500")         // Maximum number of records returned in a single poll

// Reliability settings
props.put("enable.auto.commit", "false")      // Let Deephaven manage offsets
props.put("auto.offset.reset", "earliest")    // Start from earliest available offset if no committed offset

// Security settings (uncomment and configure as needed)
// props.put("security.protocol", "SSL")
// props.put("ssl.truststore.location", "/path/to/truststore.jks")
// props.put("ssl.truststore.password", "${KAFKA_TRUSTSTORE_PASSWORD}")

// Deephaven specific settings
props.put("deephaven.offset.column.name", "Offset")
props.put("deephaven.timestamp.column.name", "Timestamp")

// Create DIS with previously configured name and storage path.
// The path must match the 'Mount Path' value entered in the Settings tab.
dis = DataImportServerTools.getDisByNameWithStorage(kafkaDisName, "/dataImportServers/iwd-kafka1")

final KafkaTableWriter.Options opts = new io.deephaven.enterprise.kafkawriter.KafkaTableWriter.Options()
opts.dataImportServer(dis)
opts.tableName(targetTable).namespace(targetNamespace)
opts.topic(kafkaTopicName)
opts.kafkaProperties(props)
opts.keySpec(io.deephaven.kafka.KafkaTools.Consume.simpleSpec("Key", String.class))
opts.valueSpec(io.deephaven.kafka.KafkaTools.Consume.simpleSpec("Value", String.class))

// Configure fixed partitioning
opts.partitionValue(today())

KafkaTableWriter.consumeToDis(opts)
```

## Advanced Configuration Options

### Partitioning Strategies

The example script uses fixed partitioning with `opts.partitionValue(today())`, which is simple but may not be optimal for all scenarios. Consider these alternative approaches:

#### Fixed Partitioning

Fixed partitioning assigns a single partition for the life of the ingester:

```groovy
// Using a date as partition
opts.partitionValue(today())

// Using a string as partition
opts.partitionValue("static_partition")
```

#### Dynamic Partitioning

Dynamic partitioning determines partitions as a function of the data, useful for time-series data:

```groovy
import io.deephaven.enterprise.kafkawriter.TimePartitionRotation
import java.time.ZoneId

// Partition by day using the KafkaTimestamp column
opts.dynamicPartitionFunction(
    "KafkaTimestamp",
    TimePartitionRotation.daily(ZoneId.of("UTC"), 7 * 24) // Keep 7 days worth of partitions
)
```

For more details on partitioning strategies, see the [Core+ Kafka documentation](../../data-guide/streaming/coreplus-kafka.md#fixed-partitions).

### Security Considerations

#### Kafka Authentication and Encryption

When connecting to a secured Kafka cluster, you'll need to configure appropriate security settings:

```groovy
// For SSL/TLS
props.put("security.protocol", "SSL")
props.put("ssl.truststore.location", "/path/to/truststore.jks")
props.put("ssl.truststore.password", "${KAFKA_TRUSTSTORE_PASSWORD}")

// For SASL authentication (e.g., PLAIN, SCRAM, GSSAPI)
props.put("security.protocol", "SASL_SSL")
props.put("sasl.mechanism", "PLAIN")
props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"user\" password=\"password\";")
```

#### Kubernetes Secrets

Instead of hardcoding credentials, use Kubernetes secrets:

1. Create a secret containing Kafka credentials:

   ```shell
   kubectl create secret generic kafka-credentials \
     --from-literal=username=user \
     --from-literal=password=password \
     -n deephaven
   ```

2. Mount the secret to your Persistent Query pod and reference environment variables in your script.

### Monitoring Kafka Ingestion

Monitoring Kafka ingestion in Kubernetes environments is important for ensuring reliability. Here are several approaches:

#### 1. Deephaven Query Monitor

Use the Deephaven Query Monitor to check the status of your Persistent Query and view any error messages.

#### 2. Kubernetes-native Monitoring

- **Pod Metrics**: Monitor CPU, memory usage of the worker pod running your ingestion
- **Pod Logs**: Check logs for ingestion-related messages
  ```shell
  kubectl logs -f deploy/merge-worker -n deephaven
  ```
- **Pod Events**: Watch for pod-related events
  ```shell
  kubectl get events -n deephaven --field-selector involvedObject.name=merge-worker-xxxx
  ```

#### 3. Prometheus & Grafana

If your Kubernetes cluster has Prometheus and Grafana installed, create dashboards for:

- JVM metrics from your Deephaven worker pods
- Kafka consumer lag metrics
- PVC storage utilization

## Troubleshooting

### Common Issues and Solutions

#### Ingester Initialization Failures

If the ingester Persistent Query fails to initialize, view the error in the **Summary** tab of the Persistent Query details. Common causes include:

1. **Schema Mismatch**: Missing or wrong-type columns between ingester configuration and table schema

   - Solution: Ensure your schema matches the data format from Kafka

2. **Kafka Connectivity Issues**:

   - Solution: Verify the Kafka broker is accessible from the Kubernetes cluster and check network policies

3. **Resource Constraints**: Pod out of memory or CPU limits exceeded

   - Solution: Increase the resource allocation for your Persistent Query pod

4. **Storage Issues**: PVC mounting problems or insufficient storage
   - Solution: Check PVC status and storage class compatibility

#### Schema Evolution Problems

When schema changes cause ingestion failures because previously written data doesn't match the new schema:

##### Option 1: Attach PVC to Code Studio

1. Launch a new Code Studio, select a merge worker, and click `Show Advanced`
2. Enter the name of the PVC used for the Kafka ingester, along with storage class name, mount point, and size
3. Launch the Code Studio
4. Delete non-matching paths using Python/Groovy:
   ```groovy
   import com.illumon.util.files.FileHelper
   FileHelper.deleteRecursivelyOnNFS(new File("/dataImportServers/iwd-kafka1/path/to/table"))
   ```

##### Option 2: Delete and Recreate PVC

1. Identify the PVC:

   ```shell
   kubectl get pvc kafka-iwd-pvc --namespace deephaven
   ```

2. Delete the PVC and its associated PV:

   ```shell
   kubectl delete pvc kafka-iwd-pvc --namespace deephaven
   kubectl delete pv pvc-0255bf9a-28c8-4fa4-a0aa-de23f05834e0
   ```

3. Restart the Persistent Query to create a new PVC

#### Debugging Connection Issues

To verify Kafka connectivity from within the Kubernetes cluster:

```shell
# Deploy a test pod with Kafka tools
kubectl run kafka-debug --image=confluentinc/cp-kafka:6.1.1 -it --rm --namespace deephaven -- bash

# Test connection to Kafka broker
kafka-broker-list.sh bootstrap-servers=kafka-broker-example.com:9092 describe
```

### Advanced Troubleshooting

For more complex issues, consider these approaches:

1. **Enable Verbose Logging**: Add logging properties to your Kafka consumer configuration

   ```groovy
   props.put("deephaven.log.level", "DEBUG")
   ```

2. **Analyze Pod State**: Generate a diagnostics dump for deeper inspection

   ```shell
   kubectl exec -it deploy/merge-worker -n deephaven -- jcmd 1 GC.heap_dump /tmp/heap.hprof
   kubectl cp deephaven/merge-worker:/tmp/heap.hprof ./heap.hprof
   ```

3. **Consult Support**: For persistent issues, contact Deephaven support with:
   - PQ logs
   - Kafka topic metadata
   - Kubernetes cluster information

## Related documentation

- [Core+ Kafka Integration](../../data-guide/streaming/coreplus-kafka.md)
- [Kafka Crash Course](../../crash-course/data-in/streaming-kafka.md)
- [Data Import Server (DIS)](../../data-guide/dis.md)
- [Kubernetes Quickstart](../../quickstarts/kubernetes-quickstart.md)
- [Kubernetes Configuration Settings](../../sys-admin/kubernetes/kubernetes-configuration-settings.md)
- [Kubernetes IAP Integration](../../sys-admin/kubernetes/kubernetes-iap-integration.md)
- [Tables and Schemas](../../data-guide/tables-and-schemas.md)
- [Legacy Kafka Integration](../../legacy/importing-data/kafka.md)
- [Apache Kafka Documentation](https://kafka.apache.org/documentation/)
- [Kubernetes Storage](https://kubernetes.io/docs/concepts/storage/)
- [Prometheus Monitoring](https://prometheus.io/docs/introduction/overview/)