1

I am trying to connect to my local kafka. I created a topic "test_producer" and created a producer for it using:

bin/kafka-console-producer.sh --topic test_producer --bootstrap-server localhost:9092

Now i am trying to connect this using node-kafka in my node application. Normal consumer is working and receiving messages but if i create a consumer group it is throwing request timeout exception

Consumer group code:

import { ConsumerGroup } from "kafka-node";
const options = {
    kafkaHost: "127.0.0.1:9092",
    batch: undefined,
    ssl: true,
    groupId: "ExampleGroup",
    sessionTimeout: 15000,
    protocol: ["roundrobin"],
    encoding: "utf8",
    fromOffset: "latest", // default
    commitOffsetsOnFirstJoin: true,
    outOfRangeOffset: "earliest",
    onRebalance: (isAlreadyMember, callback) => {
      callback();
    },
  };
  const consumerGroup = new ConsumerGroup(options, ["test_producer"]);
  consumerGroup.on("message", (message) => {
    console.log(`group:${message}`);
  });

But if i create consumer it is able to connect and receive messages.

Edit:

config/server.properties

############################# Server Basics #############################
broker.id=0

############################# Socket Server Settings #############################

num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

############################# Log Basics #############################

log.dirs=/home/kafka/logs
num.partitions=1
num.recovery.threads.per.data.dir=1

############################# Internal Topic Settings  #############################
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

############################# Log Flush Policy #############################

#log.flush.interval.messages=10000
#log.flush.interval.ms=1000

############################# Log Retention Policy #############################

log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000

############################# Zookeeper #############################
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=18000


############################# Group Coordinator Settings #############################

group.initial.rebalance.delay.ms=0
delete.topic.enable = true
advertised.listeners=PLAINTEXT://127.0.0.1:9092
listeners = PLAINTEXT://127.0.0.1:9092
alan trevor
  • 23
  • 7

1 Answers1

0

offsets.topic.replication.factor defaults to 3, and you won't be able to use consumer group features without the offsets topic.

Edit your broker config to set it to 1, and restart Kafka.

Also, you don't need port and advertised.host.name ; both are deprecated properties

OneCricketeer
  • 179,855
  • 19
  • 132
  • 245
  • I have updated the server.properties. It is sill not connecting. – alan trevor Sep 17 '22 at 19:08
  • Did the `__consumer_offsets` topic actually get created? – OneCricketeer Sep 17 '22 at 21:56
  • yes. it is listed in the topics created. bin/kafka-topics.sh --list --bootstrap-server localhost:9092 -> this command lists : __consumer_offsets, test_producer – alan trevor Sep 18 '22 at 14:37
  • Kafka doesn't use `ssl`, since your listeners are plaintext, so have you tried disabling that? And assuming you want to consume existing data, so setting `fromOffset` to earliest? – OneCricketeer Sep 18 '22 at 15:15
  • Yes. tried keeping ssl as false and changed fromOffset as earliest. Still its not connecting. But if i connect via consumer it is working. – alan trevor Sep 18 '22 at 15:20
  • Not sure, then. I suggest you use a different library that's maintained where you can get better support from the developer https://github.com/SOHU-Co/kafka-node/issues/1445 – OneCricketeer Sep 18 '22 at 15:22
  • That is the package i am using too. – alan trevor Sep 18 '22 at 15:31
  • 1
    Yes, I know. It's not maintained. So if you're "new to Kafka" or "trying to use Nodejs with Kafka", i suggest you use an alternative library suggested in that thread since that one in particular has had no new commits in years – OneCricketeer Sep 18 '22 at 15:37