我正在尝试使用 Kafka connect 使用独立模式写入数据。我将数据写入的主题是具有多个分区。但是,数据仅写入其中一个分区。当我启动多个消费者控制台时,数据仅打印到其中一个。另一个消费者控制台只有在第一个控制台关闭后才能获取任何数据。我无法弄清楚需要在配置文件中进行哪些更改才能使其写入多个分区。
这是standalone.properties
bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=1000
rest.port=8084
Run Code Online (Sandbox Code Playgroud)
连接文件源.properties:
name=local-file-source
connector.class=FileStreamSource
tasks.max=1
file=test4.txt
topic=consumer_group
Run Code Online (Sandbox Code Playgroud)
现在我使用以下命令来运行连接器:
bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties
Run Code Online (Sandbox Code Playgroud)
使用以下命令启动消费者控制台:
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic consumer_group --from-beginning --consumer-property group.id=new-consumer-group
Run Code Online (Sandbox Code Playgroud)
它只将数据打印到消费者控制台之一。但是,如果我使用生产者控制台而不是 Kafka 连接来编写消息,那么我可以看到多个消费者上的消息(以循环方式),应该是这样。但是使用Kafka connect,它只是将所有数据写入单个分区,同一组中的其他消费者必须闲置。需要更改什么才能写入循环系统中的所有分区?