SimpleExample.java

package com.blackberry.bdp.kafka.sctest;

import java.util.ArrayList;
import java.util.List;
import kafka.javaapi.PartitionMetadata;
import kafka.javaapi.TopicMetadata;
import kafka.javaapi.TopicMetadataRequest;
import kafka.javaapi.TopicMetadataResponse;
import kafka.javaapi.consumer.SimpleConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SimpleExample {

	private static final Logger LOG = LoggerFactory.getLogger(Main.class);
	
	public static void main(String[] args) {

		List<String> topicStrings = new ArrayList<>();
		topicStrings.add(args[1]);

		LOG.info("seed brokers: {}", args[0]);

		for (String seed : args[0].split(",")) {
			String seedHost = seed.split(":")[0];
			int seedPort = Integer.parseInt(seed.split(":")[1]);
			LOG.info("Trying broker @ {}:{}", seedHost, seedPort);
			SimpleConsumer consumer = null;
			try {
				consumer = new SimpleConsumer(seedHost, seedPort, 100000, 64 * 1024, "simple-consumer");
				TopicMetadataRequest req = new TopicMetadataRequest(topicStrings);
				TopicMetadataResponse resp = consumer.send(req);

				for (TopicMetadata topicMetadata : resp.topicsMetadata()) {
					for (PartitionMetadata pmd : topicMetadata.partitionsMetadata()) {
						if (pmd.replicas().size() != pmd.isr().size()) {
							LOG.error("topic: {}, partition: {}, replicas: {}, isr: {}",
								 topicMetadata.topic(),
								 pmd.partitionId(),
								 pmd.replicas(),
								 pmd.isr());
						}
					}
				}
				break; // Don't iterate onto the next broker, we're done
			} catch (Exception e) {
				LOG.error("Error getting meta data", e);
			} finally {
				if (consumer != null) {
					consumer.close();
				}
			}
		}
	}

}