Page History

import kafka.api.*;
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 offsetChannel;
// suppose a query channel has been established to some broker
try {
  queryChannel.send(new ConsumerMetadataRequest("myconsumergroup"));
  ;
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(MY_GROUP, ConsumerMetadataRequest.CurrentVersion(), correlationId++, MY_CLIENTID));
        ConsumerMetadataResponse metadataResponse = ConsumerMetadataResponse.readFrom(queryChannelchannel.readreceive().buffer());

        if (metadataResponse.errorCode() == ErrorMapping.NoError &&()) {
            Broker offsetManager = metadataResponse.coordinator != null) {
();
            // if the (metadataResponse.coordinator.host == queryChannel.host && metadataResponse.port == queryChannel.port)
coordinator is different, from the above channel's host then reconnect
          offsetChannel = queryChannelchannel.disconnect();
    else  {
      offsetChannelchannel = new BlockingChannel(coordinatoroffsetManager.host(), coordinatoroffsetManager.port(),
                                          BlockingChannel.UseDefaultBufferSize(),
                                          BlockingChannel.UseDefaultBufferSize(), socketTimeoutMs);

                                          5000 /* read timeout in millis */);
            queryChannelchannel.disconnectconnect();
      }
  } else {
            // retry the query (after backoff)
        }
    }
    catch (IOException e) {
        // retry the query (after backoff)
    }

// How to commit offsets

        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("myconsumergroup", offsets, correlationId, clientId);
try {
  offsetChannel.send(ocRequest);
  OffsetCommitResponse ocResponse
                MY_GROUP,
                offsets,
                correlationId++,
                MY_CLIENTID,
                (short) 1 /* version */); // version 1 and above commit to Kafka, version 0 commits to ZooKeeper
        try {
            channel.send(commitRequest.underlying());
            OffsetCommitResponse commitResponse = OffsetCommitResponse.readFrom(offsetChannelchannel.readreceive().buffer());
            if (commitResponse.hasError()) {
				for (partitionErrorCode: ocResponsecommitResponse.commitStatuserrors().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 partitions = new ArrayList<TopicAndPartition>();
// populate partitionList with list of partitions for which we want to look up last committed consumer offset
OffsetFetchRequest ofRequest partitions.add(testPartition0);
        OffsetFetchRequest fetchRequest = new OffsetFetchRequest(groupId, partitionList);
try {
  offsetChannel.send(ofRequest);
  OffsetFetchResponse ofResponse(
                MY_GROUP,
                partitions,
                (short) 1 /* version */, // version 1 and above fetch from Kafka, version 0 fetches from ZooKeeper
                correlationId,
                MY_CLIENTID);
		try {
	        channel.send(fetchRequest.underlying());
    	    OffsetFetchResponse fetchResponse = OffsetFetchResponse.readFrom(offsetChannelchannel.readreceive().buffer());
	     Short errorCode   OffsetMetadataAndError result = ofResponsefetchResponse.offsets.values.iterator.next()().get(testPartition0);
    	    short offsetFetchErrorCode = result.error();
            if (errorCodeoffsetFetchErrorCode == ErrorMapping.NotCoordinatorForConsumerCode()) {
    offsetChannel				channel.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)
  			} else {
	    	    long retrievedOffset = result.offset();
		        String retrievedMetadata = result.metadata();
			}
		}
		catch (IOException ioee) {
  offsetChannel			channel.disconnect();
  			// Go to step 1 and then retry the offset fetch after backoff
		}