Broker throws SchemaException: Error reading field 'topics'

[ismriv]

Hi @oleksiyk, I've tried to run a producer + group consumer, but I'm running into some problems with the consumer. Using the same setup, the simple consumer works and I'm able to receive messages.

The error in question is:

[2016-02-03 23:04:34,155] ERROR Closing socket for /172.17.42.1 because of error (kafka.network.Processor)
org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'topics': java.nio.BufferUnderflowException
    at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:66)
    at org.apache.kafka.common.requests.JoinGroupRequest.parse(JoinGroupRequest.java:85)
    at kafka.api.JoinGroupRequestAndHeader$.readFrom(JoinGroupRequestAndHeader.scala:29)
    at kafka.api.RequestKeys$$anonfun$12.apply(RequestKeys.scala:50)
    at kafka.api.RequestKeys$$anonfun$12.apply(RequestKeys.scala:50)
    at kafka.network.RequestChannel$Request.<init>(RequestChannel.scala:50)
    at kafka.network.Processor.read(SocketServer.scala:450)
    at kafka.network.Processor.run(SocketServer.scala:340)
    at java.lang.Thread.run(Thread.java:745)

What I've done to setup the environment is:

$ docker run -d --name zookeeper jplock/zookeeper:3.4.6
$ docker run -d --name kafka --link zookeeper:zookeeper ches/kafka

$ ZK_IP=$(docker inspect --format '{{ .NetworkSettings.IPAddress }}' zookeeper)
$ KAFKA_IP=$(docker inspect --format '{{ .NetworkSettings.IPAddress }}' kafka)

$ docker run --rm ches/kafka kafka-topics.sh --create --topic test --replication-factor 1 --partitions 1 --zookeeper $ZK_IP:2181
Created topic "test".

My producer is:

var Kafka = require('no-kafka');
var producer = new Kafka.Producer({
  connectionString: process.env.KAFKA_IP + ':9092'
});

return producer.init().then(function(){
  return producer.send({
      topic: 'test',
      partition: 0,
      message: {
          value: 'Hello!'
      }
  });
}).then(console.log).then(process.exit);

And my group consumer:

var Kafka = require('no-kafka');
var consumer = new Kafka.GroupConsumer({
  connectionString: process.env.KAFKA_IP + ':9092'
});
var strategies = {
    strategy: 'TestStrategy',
    subscriptions: ['test']
};

consumer.on('data', function (messageSet, topic, partition) {
    messageSet.forEach(function (m) {
        console.log(topic, partition, m.offset, m.message.value.toString('utf8'));
        // process each message and commit its offset
        consumer.commitOffset({topic: topic, partition: partition, offset: m.offset, metadata: 'optional'});
    });
});

return consumer.init(strategies).then(function() {
  // all done, now wait for messages in event listener
});

Running the consumer throws the following error: NoKafkaConnectionError (172.17.0.45:9092): Kafka server [172.17.0.45:9092] has closed connection, which is due to the above error taken from the Kafka error log file.

Any help or ideas would be appreciated!

[oleksiyk]

Kafka and no-kafka versions?

[oleksiyk]

ches/kafka seems to install Kafka 0.8.1.1
no-kafka uses V2 group membership which is only available in Kafka 0.9.

Please confirm Kafka version

[ismriv]

My bad, meant to install greytip/kafka (https://hub.docker.com/r/greytip/kafka/) but in the description it's still referencing ches/kafka. I'm running now the correct version of Kafka, that is 0.9, and no-kafka@1.2.1, the above error is gone, but I'm getting this:

Error: Failed to join the group
    at tryJoinGroup (/home/vagrant/altorepos/new-kafka/node_modules/no-kafka/lib/group_consumer.js:76:19)
    at /home/vagrant/altorepos/new-kafka/node_modules/no-kafka/lib/group_consumer.js:82:24
    at Timer.listOnTimeout [as ontimeout] (timers.js:112:15)
From previous event:
    at /home/vagrant/altorepos/new-kafka/node_modules/no-kafka/lib/group_consumer.js:81:40

[oleksiyk]

This error is only triggered when Kafka server returns 'GroupCoordinatorNotAvailable'. Are you running Kafka 0.9 with the same Zookeeper instance that was used by Kafka 0.8?

[ismriv]

I used the same Zookeeper instance, but a different new topic name test-2. I've just tried with a fresh instance of both Zookeeper & Kafka and it worked fine. But two questions:

1. Why was Kafka returning GroupCoordinatorNotAvailable in my second test using a new topic?
2. How would you go on migrating from 0.8.x to 0.9? My idea was to (somehow) do a rolling upgrade of Kafka, keep the same Zookeeper instance, and probably rename the consumer group ids.

[oleksiyk]

1. Kafka 0.9 uses incompatible with 0.8 Zookeeper structure
2. Really not sure how to migrate, this should probably be covered by Kafka docs, or maybe you can find answers in Kafka mailing list

In order to use the same Zookeeper instance for both 0.8 and 0.9 running at the same time you should run Kafka 0.9 with a different Zookeeper root:

config/server.properties:

zookeeper.connect=localhost:2181/kafka0.9

In this case all Kafka cli operations with 0.9 should also specify this root:

kafka-topics.sh --zookeeper 127.0.0.1:2181/kafka0.9 --create --topic kafka-test-topic --partitions 3 --replication-factor 1

[ismriv]

Thanks @oleksiyk!

[FreeSlaver]

good job