# Kafka Consumer Group

The Kafka consumer group uses external Kafka as a message queue, which can convert consumer messages from Kafka into MQTT messages and publish them in emqx.

Set up the Kafka environment, taking MacOS X as an example:

wget https://archive.apache.org/dist/kafka/2.8.0/kafka_2.13-2.8.0.tgz

tar -xzf  kafka_2.13-2.8.0.tgz

cd kafka_2.13-2.8.0

# Start Zookeeper
$ ./bin/zookeeper-server-start.sh config/zookeeper.properties
# Start Kafka
$ ./bin/kafka-server-start.sh config/server.properties

1
2
3
4
5
6
7
8
9
10

:::

Kafka consumer groups do not support Kafka versions below 0.9

Before creating resources, you need to create Kafka topics in advance, otherwise an error will be prompted

:::

Create Kafka topic:

./bin/kafka-topics.sh --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testTopic --create

# Create module

Open EMQX Dashboard (opens new window), click the "Modules" tab on the left, and choose to add:

Select Kafka consumer group module:

Click "Select", and then select "Kafka Consumer Group":

Fill in the relevant parameters:

1). Kafka server address

2). Kafka consumer connection pool size

3). Kafka subscription topic

4). MQTT message subject

5). MQTT topic service quality

6). Kafka Max Bytes (the maximum number of bytes of messages consumed each time from Kafka)

7). Kafka Offset Reset Policy (reset Offset policy, reset_to_latest | reset_by_subdcriber)

Kafka server address
Kafka consumer connection pool size
Kafka subscription topic
MQTT message subject
MQTT theme service quality
MQTT Payload. Use Kafka message.value or entire message
Binary Key encode mode, force UTF-8 or base64 encode. The encoding method of the key in the message, if the key value is a non-string or a value that may generate a character set encoding exception, base64 mode is recommended
Binary Value encode mode, force UTF-8 or base64 encode. The encoding method of the value in the message, if the value is a non-string or a value that may generate a character set encoding exception, base64 mode is recommended
Kafka Max Bytes (the maximum number of bytes of messages consumed each time from Kafka)
Kafka Offset Reset Policy (reset Offset policy, reset_to_latest | reset_by_subdcriber)
Is Kafka consumer reconnected?

After clicking Add, the module is added:

The resource has been created, now use Dashboard's websocket tool to subscribe to the MQTT topic "TestTopic":

Use the kafka command line to produce a message:

./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic TestTopic

The websocket tool of Dashboard received the message "hello-kafka" produced by Kafka:

← Stomp Gateway Pulsar Consumer Group →