External System Integration #

In large-scale application ecosystems, Apache Kafka is often surrounded by various database technologies, search engines, caching systems, and other message brokers working synergistically together. To support business needs, data entering Kafka doesn’t just sit there; that data must be instantly indexed into Elasticsearch for full-text search, synchronized to Redis for microsecond-latency caching, or even integrated with legacy message brokers like RabbitMQ or ActiveMQ during system migration periods. To realize this multi-system integration reliably and safely, we leverage Kafka Connect. This article will dissect Kafka Connect integration architecture with Elasticsearch and Redis, review migration strategies from JMS/AMQP message brokers, and discuss credential management and authentication best practices in production.

Elasticsearch/OpenSearch Integration: Real-Time Indexing #

Elasticsearch (and its open-source equivalent, OpenSearch) is a document-oriented database that’s very fast for full-text data searches and log analysis. Connecting Kafka with Elasticsearch through the Elasticsearch Sink Connector lets us build robust real-time indexing pipelines.

flowchart LR
    subgraph KafkaCluster["Kafka Cluster"]
        Topic(("orders Topic"))
    end

    subgraph ConnectWorker["Connect Worker Node"]
        ESSink["Elasticsearch Sink Task"]
        BulkBuffer["Bulk Request Buffer"]
    end

    subgraph ESCluster["Elasticsearch Cluster"]
        ESIndex[/"orders Index"\]
    end

    Topic -->|1. Pull Record| ESSink
    ESSink -->|"2. Buffer (Size: 1000)"| BulkBuffer
    BulkBuffer -->|3. POST _bulk API| ESIndex

1. Maintaining Document Idempotency (key.ignore) #

In production, temporary network failures can cause Kafka Connect to send the same message more than once (at-least-once delivery). If not anticipated, these resends produce duplicate documents in Elasticsearch.

  • Solution: We must set the "key.ignore": "false" property. This setting tells the Sink Connector to take the message key from Kafka and use it directly as the document ID (_id) in Elasticsearch. If the same message is resent, Elasticsearch only does an upsert operation on the existing document with that ID, not creating a new document.

2. Batching Optimization (batch.size & flush.timeout.ms) #

Sending documents to Elasticsearch one by one via HTTP requests is a performance killer. We must leverage the Elasticsearch Bulk API to send hundreds or thousands of documents at once in one HTTP POST request.

  • batch.size: The maximum number of documents collected in the task buffer before sending via the Bulk API (Example: set to 1000 or 2000 documents).
  • flush.timeout.ms: The maximum wait limit (in milliseconds) before the buffer is sent even if the document count hasn’t reached batch.size (Example: 10000 ms or 10 seconds). This is important to prevent data stuck in the buffer when Kafka’s data rate is quiet.

Error Handling and Reconciliation Details on the Elasticsearch Sink #

When operating a data integration pipeline to Elasticsearch, there are several typical production errors we must mitigate so the Sink task doesn’t suddenly crash:

1. Version Conflicts (VersionConflictEngineException) #

This error happens when several task threads try to update Elasticsearch documents with the same ID but different versions simultaneously.

  • Solution: Use the "write.method": "upsert" property setting. This method instructs Elasticsearch to overwrite old fields with new fields regardless of document version numbers, or we can set "version.type": "external" and use Kafka offsets as external document version numbers to guarantee linear writes.

2. Mapping Mismatches (Mapping Mismatch Exception) #

This error happens when the data type of a field in a Kafka Connect message doesn’t match the pre-formed Elasticsearch index schema. For example, the phone_number column was originally defined as integer in the ES index, but a new Kafka record sends the string "+62-811-...".

  • Solution: Because this is a permanent error (poison pill), the query will never succeed with retries. We must route this misformatted message to a Dead Letter Queue (DLQ) using the errors.tolerance=all property so the task doesn’t crash and jam the entire pipeline.

Redis Integration: Caching & State Sync Patterns #

Redis is an in-memory data store very popular for database caching, session management, and real-time counters. Connecting Kafka with Redis via the Redis Sink Connector lets us build automatic cache synchronization systems.

Cache Integration Pattern (Write-Behind Caching) #

In traditional architectures, business applications must write data to the main database, then manually delete or update the cache in Redis (Cache-Aside). This pattern is prone to race conditions and data desynchronization.

With Kafka Connect, we implement the Write-Behind Caching pattern:

  1. Business applications only write transaction data to the main database.
  2. A CDC tool (Debezium Source) detects new transactions and sends them to Kafka.
  3. The Redis Sink Connector reads events from Kafka and immediately updates Redis data asynchronously.

This frees application code from complex cache management logic and ensures Redis always has the newest data (eventual consistency) with sub-millisecond latency.

Redis Data Structure Mapping Details #

When streaming data to Redis, we must design how Kafka Connect records map into Redis data structures:

  • Redis Strings: The simplest mapping format where the entire Kafka Connect record value payload is serialized into a JSON string or Avro/Protobuf binary bytes, then stored under one-by-one Redis keys.
    • Redis Syntax: SET customer:1024 "{\"name\":\"Rudi\",\"email\":\"[email protected]\"}"
    • Characteristics: Very efficient for whole document reads, but doesn’t support isolated individual field updates or reads.
  • Redis Hashes: Highly recommended if our Kafka records have structured schemas with many columns. Each column in the Kafka record maps to a key-value field inside a Redis Hash object.
    • Redis Syntax: HMSET customer:1024 name "Rudi" email "[email protected]" age 25
    • Characteristics: Saves memory and allows downstream applications to efficiently read or update individual fields using the HGET or HSET commands.
  • Redis Sorted Sets (ZSET): Used if we want to map queue or leaderboard data based on numeric scores (for example game scores or transaction timestamps).
    • Redis Syntax: ZADD customer:leaderboard 950 "customer:1024"

Legacy Message Queue System Integration (RabbitMQ/ActiveMQ) in Practice #

Many large companies modernizing their infrastructure want to move workloads from legacy AMQP or JMS-based message brokers to Apache Kafka. During transition periods that can last months, both brokers must exchange data seamlessly.

1. Using the JMS / AMQP Source Connector #

The JMS Source Connector acts as a consumer client on legacy brokers (like ActiveMQ). It subscribes to Queues or Topics in ActiveMQ, pulls messages, translates JMS header properties (like JMSCorrelationID, JMSReplyTo) into Kafka message headers, and sends them to Kafka.

2. Using the JMS / AMQP Sink Connector #

Conversely, the JMS Sink Connector reads data from Kafka and publishes it to RabbitMQ queues so legacy applications that haven’t been upgraded can still process that data.

Main Challenge: Data format conversion. JMS messages are often serialized Java objects (ObjectMessage) or binary maps (MapMessage). We must use the right Connect converters (like BytesConverter) and add custom transformations if message formats need cleaning before entering Kafka.

RabbitMQ Source Connector Configuration Example #

Here’s a JSON properties example for pulling data from a RabbitMQ queue into a Kafka topic:

{
  "name": "rabbitmq-source-connector",
  "config": {
    "connector.class": "com.ibm.eventstreams.connect.rabbitmq.RabbitMQSourceConnector",
    "tasks.max": "2",
    "rabbitmq.hosts": "rabbitmq-broker-1.prod.internal:5672,rabbitmq-broker-2.prod.internal:5672",
    "rabbitmq.username": "connect_importer",
    "rabbitmq.password": "${file:/etc/connect/secrets:rabbitmq_password}",
    "rabbitmq.queue": "payment-events-legacy",
    
    "//": "Determining the destination topic in Kafka",
    "kafka.topic": "legacy-payments",
    
    "//": "No Data Loss Guarantee: Only tell RabbitMQ to delete the message (ACK)",
    "//": "after the message is successfully written persistently to the Kafka broker",
    "rabbitmq.auto.ack": "false"
  }
}

JMS Sink Connector Configuration Example (ActiveMQ) #

Below is a Sink Connector configuration for sending processed Kafka data to an ActiveMQ Queue:

{
  "name": "activemq-queue-sink",
  "config": {
    "connector.class": "io.confluent.connect.jms.JmsSinkConnector",
    "tasks.max": "1",
    "topics": "approved-loans",
    "java.naming.factory.initial": "org.apache.activemq.jndi.ActiveMQInitialContextFactory",
    "java.naming.provider.url": "tcp://activemq-server:61616",
    
    "//": "Determining the target ActiveMQ queue",
    "jms.destination.type": "queue",
    "jms.destination.name": "loans.processed",
    
    "//": "Configuration of the JMS message type sent",
    "message.type": "text"
  }
}

Authentication and Connection Security Management #

Connecting Kafka Connect to various external systems in production requires implementing very strict security protocols to protect sensitive data in transit and prevent illegal access.

1. Transport Encryption (SSL/TLS) #

Always enable SSL/TLS encryption on every connection to target systems.

  • On Elasticsearch: Enable HTTPS (https://elasticsearch:9200) and register trusted Certificate Authority (CA) certificates in the Connect worker’s Java Truststore configuration.
  • On Redis: Use the secure Redis over TLS (RoT) connection.

2. Authentication Protocols #

Use modern authentication mechanisms supported by target systems:

  • Elasticsearch: Basic Authentication (Username/Password) or API Keys (recommended for service accounts).
  • Redis: The AUTH command with encrypted usernames and passwords.
  • AMQP/JMS Brokers: SASL/PLAIN authentication or SSL client certificate-based authentication (Mutual TLS / mTLS).

3. Preventing Hardcoded Credentials via Config Providers #

Writing database passwords or API Keys directly (hardcoded) inside connector JSON configuration files stored in Git repositories is a fatal security violation.

  • Solution: Kafka Connect provides the ConfigProvider framework allowing workers to dynamically read credential values from external providers when connectors run.

We can configure Connect workers to read secrets from safe local files, AWS Secrets Manager, Google Secret Manager, or HashiCorp Vault.

Example Connect worker settings for reading a local secrets file (/opt/connect/secrets.properties):

# Registering a config provider named 'file'
config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider

Inside the connector JSON configuration, we just reference those secrets using special placeholders:

{
  "name": "elasticsearch-sink-secure",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "connection.url": "https://elasticsearch-node:9200",
    "connection.username": "${file:/opt/connect/secrets.properties:es_username}",
    "connection.password": "${file:/opt/connect/secrets.properties:es_password}"
  }
}

When the Connect worker loads this JSON, it automatically replaces ${file:...:es_password} with the actual secure password from disk.

Comprehensive Elasticsearch Sink Connector Configuration Example #

Here’s a complete JSON configuration for deploying a secure production Elasticsearch Sink Connector, using API Key authentication, leveraging Kafka keys as document IDs for idempotency, and equipped with bulk buffer optimization:

{
  "name": "elasticsearch-audit-sink",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "4",
    "topics": "production-audit-logs",
    
    "//": "--- Secure HTTPS Connection Details ---",
    "connection.url": "https://es-secure-cluster:9200",
    "connection.username": "connect_writer",
    "connection.password": "${file:/etc/connect/secrets:es_writer_password}",
    
    "//": "Ignoring self-signed SSL verification if in a dev environment",
    "//": "DON'T set this to true in the main production environment!",
    "connection.ssl.truststore.location": "/etc/connect/kafka.connect.truststore.jks",
    "connection.ssl.truststore.password": "${file:/etc/connect/secrets:jks_password}",
    
    "//": "--- Index Settings and Idempotency ---",
    "//": "✓ CORRECT: Using Kafka keys as ES document IDs to prevent duplication",
    "key.ignore": "false",
    "schema.ignore": "true",
    "write.method": "upsert",
    
    "//": "--- Bulk Request Optimization (Backpressure Protection) ---",
    "batch.size": "2000",
    "flush.timeout.ms": "5000",
    "max.buffered.records": "20000",
    "max.in.flight.requests": "5",
    
    "//": "--- Network Error Handling ---",
    "max.retry.time.ms": "60000",
    "retry.backoff.ms": "2000",
    
    "//": "--- Avro Converter ---",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081"
  }
}

Summary #

  • Elasticsearch Integration — The Elasticsearch Sink Connector provides a real-time document indexing pipeline with Bulk API optimization (batch.size and flush.timeout.ms) for high throughput.
  • Idempotency via Key — Set key.ignore=false so Kafka message keys are used as unique document IDs in Elasticsearch to prevent data duplication from resends.
  • Write-Behind Caching — Use the Kafka Connect Redis Sink to asynchronously update cache data in Redis based on database CDC events, guaranteeing eventual consistency without burdening application code.
  • Redis Mappings — Connect Kafka records with Redis Strings (whole JSON documents) or Redis Hashes (field-value pairs) according to downstream query characteristics.
  • Broker Migration — Integration with RabbitMQ or ActiveMQ is done using JMS/AMQP connectors to bridge data communication during system migration periods.
  • Credential Protection — Secure target system credentials by leveraging Kafka Connect’s built-in Config Providers so passwords aren’t hardcoded in configuration files.

← Previous: File & Object Storage Integration Next: Scaling & Resource Allocation →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact