Versions Compared

Old Version 9

changes.mady.by.user Joel Koshy

Saved on

compared with

New Version 10

changes.mady.by.user Joel Koshy

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

The official Kafka documentation describes how the feature works and how to migrate offsets from ZooKeeper to Kafka. This wiki provides sample code that shows how to use the new Kafka-based offset storage mechanism.

Step 1: Discover and connect to the offset manager for a consumer group by issuing a consumer metadata request to any broker

Code Block
import kafka.api.ConsumerMetadataRequest;

...

BlockingChannel offsetChannel;
BlockingChannel queryChannel;
// establish the query channel to some broker
try {
  queryChannel.*;
import kafka.cluster.Broker;
import kafka.common.OffsetAndMetadata;
import kafka.common.OffsetMetadataAndError;
import kafka.common.TopicAndPartition;
import kafka.javaapi.ConsumerMetadataResponse;
import kafka.javaapi.OffsetCommitRequest;
import kafka.javaapi.OffsetCommitResponse;
import kafka.javaapi.OffsetFetchRequest;
import kafka.javaapi.OffsetFetchResponse;
import kafka.network.BlockingChannel;
import java.util.*;

...

    try {
        BlockingChannel channel = new BlockingChannel("localhost", 9092,
                BlockingChannel.UseDefaultBufferSize(),
                BlockingChannel.UseDefaultBufferSize(),
                5000 /* read timeout in millis */);
        channel.connect();
        final String MY_GROUP = "demoGroup";
        final String MY_CLIENTID = "demoClientId";
        int correlationId = 0;
        final TopicAndPartition testPartition0 = new TopicAndPartition("demoTopic", 0);
        final TopicAndPartition testPartition1 = new TopicAndPartition("demoTopic", 1);
        channel.send(new ConsumerMetadataRequest(consumerGroupStringMY_GROUP, ConsumerMetadataRequest.CurrentVersion(), correlationId++, clientIdStringMY_CLIENTID));
        ConsumerMetadataResponse metadataResponse = ConsumerMetadataResponse.readFrom(queryChannelchannel.receive().buffer());

        if (metadataResponse.errorCode() == ErrorMapping.NoError()) &&{
            Broker offsetManager = metadataResponse.coordinator() != null) {
;
            // if the (metadataResponse.coordinator().host() == queryChannel.host() && metadataResponse.coordinator().port() == queryChannel.port)
coordinator is different, from the above channel's host then reconnect
          offsetChannel = queryChannelchannel.disconnect();
     else {
      offsetChannelchannel = new BlockingChannel(metadataResponse.coordinator()offsetManager.host(), metadataResponseoffsetManager.coordinator().port(),
                                          BlockingChannel.UseDefaultBufferSize(),
                                          BlockingChannel.UseDefaultBufferSize(), socketTimeoutMs
                                          5000 /* read timeout in millis */);
       queryChannel.disconnect     channel.connect();
    }
    } else {
            // retry the query (after backoff)
        }
    }
    catch (IOException e) {
        // retry the query (after backoff)
    }

Step 2: Issue the OffsetCommitRequest or OffsetFetchRequest to the offset manager

Code Block
// How to commit offsets

import kafka.javaapi.OffsetCommitRequest;

...

        long now = System.currentTimeMillis();
        Map<TopicAndPartition, OffsetAndMetadata> offsets;
// populate the offsets map
OffsetCommitRequest ocRequest = new LinkedHashMap<TopicAndPartition, OffsetAndMetadata>();
        offsets.put(testPartition0, new OffsetAndMetadata(100L, "associated metadata", now));
        offsets.put(testPartition1, new OffsetAndMetadata(200L, "more metadata", now));
        OffsetCommitRequest commitRequest = new OffsetCommitRequest(consumerGroupString, offsets, correlationId, clientIdString, 
                MY_GROUP,
                offsets,
                correlationId++,
                MY_CLIENTID,
                (short) 1 /* version */); /* note that// version 1 and above commit to Kafka, version 0 commits offsets to ZooKeeper */

        try {
  offsetChannel          channel.send(ocRequestcommitRequest.underlying());
            OffsetCommitResponse ocResponsecommitResponse = OffsetCommitResponse.readFrom(offsetChannelchannel.receive().buffer());
            if (commitResponse.hasError()) {
				for (partitionErrorCode: ocResponsecommitResponse.errors().values()) {
				    if (partitionErrorCode == ErrorMapping.OffsetMetadataTooLargeCode()) {
			  	        // You must reduce the size of the metadata if you wish to retry
				    } else if (partitionErrorCode == ErrorMapping.NotCoordinatorForConsumerCode() || partitionErrorCode == ErrorMapping.ConsumerCoordinatorNotAvailableCode()) {
				      offsetChannelchannel.disconnect();
                      // Go to step 1 (offset manager has moved) and then retry the commit to the new offset manager
 				    } else {
				        error("Error in commit");
 // log and retry the commit
			        }
 	     // retry the commit
    }
  			}
		}
		catch (IOException ioe) {
	     offsetChannel   channel.disconnect();
		    // Go to step 1 and then retry the commit
} 

	    }

// How to fetch offsets

List<TopicAndPartition> partitionList;
// populate partitionList with list of partitions for which we want to look up last committed consumer offset
OffsetFetchRequest ofRequest = new OffsetFetchRequest(consumerGroupString, partitionList);
try {
  offsetChannel.send(ofRequest);
  OffsetFetchResponse ofResponse = OffsetFetchResponse.readFrom(offsetChannel.receive().buffer());
  Short errorCode = ofResponse.offsets.values.iterator.next().error();
  if (errorCode == ErrorMapping.NotCoordinatorForConsumerCode()) {
    offsetChannel.disconnect();
    // Go to step 1 (offset manager has moved) and then retry the offset fetch to the new offset manager
  } else if (errorCode == ErrorMapping.OffsetsLoadInProgress()) {
    // retry the offset fetch (after backoff)
  }
}
catch (IOException ioe) {
  offsetChannel.disconnect();
  // Go to step 1 and then retry the offset fetch
}