Kafka Connector built on top of Connect SDK

Get Started
Functionality
Learning
Extension
Dev Tooling
APIs
Releases

How-to Guide

Author:

Fluent Commerce

Changed on:

30 Oct 2023

Key Points

The User can configure Kafka Connector on top of the Connect SDK.
Some configuration is required.

Steps

Kafka Connector built on top of Connect SDK

We have extended our queue listener support beyond SQS and now also includes Kafka now.Kafka and SQS can be used interchangeably as queue listeners, providing more options to our users.Provided reference solutions can be used to demonstrate how to apply these changes to configuration files and use either queue listener. With these examples, it's easy to switch between SQS or Kafka, depending on your specific needs and requirements.Follow this Fluent Connector Configuration | Kafka Listener Configuration for a more detailed explanation of Kafka configuration.

Prerequisites

At least a running Kafka broker, ZooKeeper container, Kafdrop(Optional)
localstack container is running
Access to a Fluent account
API key/secret for communicating with external network

Create a Docker file `docker-compose.yml` with content below to quickly set up a Kafka broker.

1version: "3.9"
2services:
3  localstack:
4    container_name: localstack
5    image: localstack/localstack
6    restart: unless-stopped
7    ports:
8      - "4510-4559:4510-4559"  # external service port range
9      - "4566:4566"
10    environment:
11      - AWS_DEFAULT_REGION=us-east-1
12      - SERVICES=sqs,secretsmanager
13      - HOSTNAME=localstack
14      - HOSTNAME_EXTERNAL=localstack
15    volumes:
16      - ./docker/localstack:/docker-entrypoint-initaws.d
17
18  kafdrop:
19    image: obsidiandynamics/kafdrop
20    container_name: kafkadrop
21    restart: "unless-stopped"
22    ports:
23      - "9000:9000"
24    environment:
25      KAFKA_BROKERCONNECT: "kafka:29092"
26      JVM_OPTS: "-Xms16M -Xmx48M -Xss180K -XX:-TieredCompilation -XX:+UseStringDeduplication -noverify"
27    depends_on:
28      - "kafka"
29
30  zookeeper:
31    image: wurstmeister/zookeeper
32    ports:
33      - "2181:2181"
34    restart: unless-stopped
35
36  kafka:
37    image: wurstmeister/kafka
38    ports:
39      - "9092:9092"
40    environment:
41      KAFKA_ADVERTISED_HOST_NAME: kafka
42      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
43      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
44      KAFKA_LISTENERS: "INTERNAL://:29092,EXTERNAL://:9092"
45      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka:29092,EXTERNAL://localhost:9092"
46      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
47      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"
48      KAFKA_ZOOKEEPER_SESSION_TIMEOUT: "6000"
49      KAFKA_RESTART_ATTEMPTS: "10"
50      KAFKA_RESTART_DELAY: "5"
51    restart: unless-stopped
52    depends_on:
53      - zookeeper

Configuration Changes

This is an example of using Kafka:

1fluent-connect:
2  connector-name: ${connectorName}
3  listeners:
4  # Internal listener:
5    messages:
6      name: "KAFKA_EVENTS"
7      type: "kafka"
8      retry: 5
9      props:
10        retryIntervalInSec: 10
11    batch:
12      name: "KAFKA_BATCH"
13      retry: 1
14      type: "kafka"
15      props:
16        retryIntervalInSec: 10
17    notification:
18      name: "KAFKA_NOTIFICATION"
19      retry: 5
20      type: "sqs"
21      poll-wait: 1
22      props:
23        retryIntervalInSec: 10
24kafka:
25  props:
26    bootstrap.servers: localhost:9092

Property	Mandatory	Description	Default Value
`fluent-connect.listeners.<listener_name>.type`	Yes		`"sqs"` or `"kafka"`
`fluent-connect.listeners.<listener_name>.retry`	No	The number of retries for a topic determines the number of corresponding retry topics that will be created. These retry topics should follow a common naming convention:`{listenerId}-retry-{iterationCount}`For instance, if the "messages" listener has been configured with three retries, the Kafka container should include the following topics:messagesmessages-retry-1messages-retry-2messages-retry-3 (Maximum retry is 5. Last failed retry will be moved to Dead Letter Queue topic.)	n/a
`fluent-connect.listeners.<listener_name>.props`:`retryIntervalInSec`	No	The delay time between retry attempts.(The delay time should not be greater than max pool interval of current Kafka broker.)	n/a
`kafka.props`	No	The connection configurations for Kafka producer and consumer can be referred to as: Producer Configs Consumer Configs (If this property is not set, the producer and consumer will connect to the Kafka on local at `localhost:9092`)	`kafka:`` props:``bootstrap.servers: localhost:9092`

`kafka.props` example:

1kafka:
2  props:
3    # Required connection configs for Kafka Local
4    bootstrap.servers: localhost:9092
5
6    # Best practice for higher availability in Apache Kafka clients prior to 3.0
7    session.timeout.ms: 45000
8    
9    # Best practice for Kafka producer to prevent data loss
10    acks: all

1kafka:
2  props:
3    # Required connection configs for Confluent Cloud
4    bootstrap.servers: pkc-6ojv2.us-west4.gcp.confluent.cloud:9092
5    security.protocol: SASL_SSL
6    sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username='{{ CLUSTER_API_KEY }}' password='{{ CLUSTER_API_SECRET }}';
7    sasl.mechanism: PLAIN
8    client.dns.lookup: use_all_dns_ips
9
10    # Best practice for higher availability in Apache Kafka clients prior to 3.0
11    session.timeout.ms: 45000
12
13    # Best practice for Kafka producer to prevent data loss
14    acks: all

Notes

The messages that are consumed by our listeners can be accessed at the Kafdrop host server.

Kafka Connector built on top of Connect SDK

Prerequisites

At least a running Kafka broker, ZooKeeper container, Kafdrop(Optional)
localstack container is running
Access to a Fluent account
API key/secret for communicating with external network

Create a Docker file `docker-compose.yml` with content below to quickly set up a Kafka broker.

1version: "3.9"
2services:
3  localstack:
4    container_name: localstack
5    image: localstack/localstack
6    restart: unless-stopped
7    ports:
8      - "4510-4559:4510-4559"  # external service port range
9      - "4566:4566"
10    environment:
11      - AWS_DEFAULT_REGION=us-east-1
12      - SERVICES=sqs,secretsmanager
13      - HOSTNAME=localstack
14      - HOSTNAME_EXTERNAL=localstack
15    volumes:
16      - ./docker/localstack:/docker-entrypoint-initaws.d
17
18  kafdrop:
19    image: obsidiandynamics/kafdrop
20    container_name: kafkadrop
21    restart: "unless-stopped"
22    ports:
23      - "9000:9000"
24    environment:
25      KAFKA_BROKERCONNECT: "kafka:29092"
26      JVM_OPTS: "-Xms16M -Xmx48M -Xss180K -XX:-TieredCompilation -XX:+UseStringDeduplication -noverify"
27    depends_on:
28      - "kafka"
29
30  zookeeper:
31    image: wurstmeister/zookeeper
32    ports:
33      - "2181:2181"
34    restart: unless-stopped
35
36  kafka:
37    image: wurstmeister/kafka
38    ports:
39      - "9092:9092"
40    environment:
41      KAFKA_ADVERTISED_HOST_NAME: kafka
42      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
43      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
44      KAFKA_LISTENERS: "INTERNAL://:29092,EXTERNAL://:9092"
45      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka:29092,EXTERNAL://localhost:9092"
46      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
47      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"
48      KAFKA_ZOOKEEPER_SESSION_TIMEOUT: "6000"
49      KAFKA_RESTART_ATTEMPTS: "10"
50      KAFKA_RESTART_DELAY: "5"
51    restart: unless-stopped
52    depends_on:
53      - zookeeper

Configuration Changes

This is an example of using Kafka:

1fluent-connect:
2  connector-name: ${connectorName}
3  listeners:
4  # Internal listener:
5    messages:
6      name: "KAFKA_EVENTS"
7      type: "kafka"
8      retry: 5
9      props:
10        retryIntervalInSec: 10
11    batch:
12      name: "KAFKA_BATCH"
13      retry: 1
14      type: "kafka"
15      props:
16        retryIntervalInSec: 10
17    notification:
18      name: "KAFKA_NOTIFICATION"
19      retry: 5
20      type: "sqs"
21      poll-wait: 1
22      props:
23        retryIntervalInSec: 10
24kafka:
25  props:
26    bootstrap.servers: localhost:9092

Property	Mandatory	Description	Default Value
`fluent-connect.listeners.<listener_name>.type`	Yes		`"sqs"` or `"kafka"`
`fluent-connect.listeners.<listener_name>.retry`	No	The number of retries for a topic determines the number of corresponding retry topics that will be created. These retry topics should follow a common naming convention:`{listenerId}-retry-{iterationCount}`For instance, if the "messages" listener has been configured with three retries, the Kafka container should include the following topics:messagesmessages-retry-1messages-retry-2messages-retry-3 (Maximum retry is 5. Last failed retry will be moved to Dead Letter Queue topic.)	n/a
`fluent-connect.listeners.<listener_name>.props`:`retryIntervalInSec`	No	The delay time between retry attempts.(The delay time should not be greater than max pool interval of current Kafka broker.)	n/a
`kafka.props`	No	The connection configurations for Kafka producer and consumer can be referred to as: Producer Configs Consumer Configs (If this property is not set, the producer and consumer will connect to the Kafka on local at `localhost:9092`)	`kafka:`` props:``bootstrap.servers: localhost:9092`

`kafka.props` example:

1kafka:
2  props:
3    # Required connection configs for Kafka Local
4    bootstrap.servers: localhost:9092
5
6    # Best practice for higher availability in Apache Kafka clients prior to 3.0
7    session.timeout.ms: 45000
8    
9    # Best practice for Kafka producer to prevent data loss
10    acks: all

1kafka:
2  props:
3    # Required connection configs for Confluent Cloud
4    bootstrap.servers: pkc-6ojv2.us-west4.gcp.confluent.cloud:9092
5    security.protocol: SASL_SSL
6    sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username='{{ CLUSTER_API_KEY }}' password='{{ CLUSTER_API_SECRET }}';
7    sasl.mechanism: PLAIN
8    client.dns.lookup: use_all_dns_ips
9
10    # Best practice for higher availability in Apache Kafka clients prior to 3.0
11    session.timeout.ms: 45000
12
13    # Best practice for Kafka producer to prevent data loss
14    acks: all

Notes

The messages that are consumed by our listeners can be accessed at the Kafdrop host server.

Kafka Connector built on top of Connect SDK

Key Points

Steps

Kafka Connector built on top of Connect SDK

Prerequisites

Configuration Changes

Fluent Commerce

Building Blocks

By Type

Helpful Resources

Quick Links

Kafka Connector built on top of Connect SDK

Prerequisites

Configuration Changes