本文絕大部分內容參考自：http://www.cnblogs.com/huxi2b/p/6061110.html和http://www.cnblogs.com/huxi2b/p/7183219.html

一、Kafka讀取__consumer_offsets

注意：該實驗受限於kafka版本，我在kafka_2.11-0.9.0.1和kafka_2.10-0.10.1.0中都成功了，而在較舊的kafka_2.10-0.8.2.0（根本就不會產生__consumer_offsets）和最新的kafka_2.11-0.11.0.0（在第6步的時候報錯Exception in thread "main" java.lang.ClassNotFoundException: kafka.coordinator.GroupMetadataManager$OffsetsMessageFormatter）中卻無法完成

眾所周知，由於Zookeeper並不適合大批量的頻繁寫入操作，新版Kafka（0.8版本之後）已推薦將consumer的位移資訊儲存在Kafka內部的topic中，即__consumer_offsets topic，並且預設提供了kafka_consumer_groups.sh指令碼供使用者檢視consumer資訊。
不過依然有很多使用者希望瞭解__consumer_offsets topic內部到底儲存了什麼資訊，特別是想查詢某些consumer group的位移是如何在該topic中儲存的。針對這些問題，本文將結合一個例項探討如何使用kafka-simple-consumer-shell指令碼來查詢該內部topic。

1.建立topic “test”
[[email protected] kafka_2.10-0.10.1.0]$ bin/kafka-topics.sh --create --zookeeper h153:2181 --replication-factor 1 --partitions 2 --topic test

2.使用kafka-console-producer.sh指令碼生產訊息(本例中生產了4條訊息)
[[email protected] kafka_2.10-0.10.1.0]$ bin/kafka-console-producer.sh --broker-list h153:9092 --topic test

3.驗證訊息生產成功
[[email protected] kafka_2.10-0.10.1.0]$ bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list h153:9092 --topic test --time -1
test:1:2
test:0:2

引數解釋：--time -1 表示從最新的時間的offset中得到資料條數
輸出結果每個欄位分別表示topic、partition、untilOffset
上面的輸出結果表明總共生產了4條訊息

4.建立一個console consumer group
[[email protected] kafka_2.10-0.10.1.0]$ bin/kafka-console-consumer.sh --bootstrap-server h153:9092 --topic test --from-beginning --new-consumer

在kafka啟動視窗你會看見輸出這些資訊
[2017-09-26 21:49:54,454] INFO [Group Metadata Manager on Broker 0]: Loading offsets and group metadata from [__consumer_offsets,32] (kafka.coordinator.GroupMetadataManager)
[2017-09-26 21:49:54,457] INFO [Group Metadata Manager on Broker 0]: Finished loading offsets from [__consumer_offsets,32] in 3 milliseconds. (kafka.coordinator.GroupMetadataManager)
[2017-09-26 21:49:54,457] INFO [Group Metadata Manager on Broker 0]: Loading offsets and group metadata from [__consumer_offsets,35] (kafka.coordinator.GroupMetadataManager)
（預設情況下__consumer_offsets有50個分割槽）

使用bin/kafka-topics.sh  --list --zookeeper h153:2181你會看到__consumer_offsets生成

5.獲取該consumer group的group id(後面需要根據該id查詢它的位移資訊)
[[email protected] kafka_2.10-0.10.1.0]$ bin/kafka-consumer-groups.sh --bootstrap-server h153:9092 --list --new-consumer
輸出： console-consumer-88985  (記住這個id！)

6.查詢__consumer_offsets topic所有內容
注意：執行下面命令前先要在consumer.properties中設定exclude.internal.topics=false否則該執行該命令後卡住不動，按Ctrl+C也無法結束

[[email protected] kafka_2.10-0.10.1.0]$ bin/kafka-console-consumer.sh --topic __consumer_offsets --zookeeper h153:2181 --formatter "kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter" --consumer.config config/consumer.properties --from-beginning
[console-consumer-88985,test,1]::[OffsetMetadata[2,NO_METADATA],CommitTime 1506433800225,ExpirationTime 1506520200225]
[console-consumer-88985,test,0]::[OffsetMetadata[2,NO_METADATA],CommitTime 1506433800225,ExpirationTime 1506520200225]
[console-consumer-88985,test,1]::[OffsetMetadata[2,NO_METADATA],CommitTime 1506433800326,ExpirationTime 1506520200326]
[console-consumer-88985,test,0]::[OffsetMetadata[2,NO_METADATA],CommitTime 1506433800326,ExpirationTime 1506520200326]
（第二次執行這個命令的時候得加--delete-consumer-offsets）

7.計算指定consumer group在__consumer_offsets topic中分割槽資訊
這時候就用到了第5步獲取的group.id(本例中是console-consumer-88985)。Kafka會使用下面公式計算該group位移儲存在__consumer_offsets的哪個分割槽上：
Math.abs(groupID.hashCode()) % numPartitions

所以在本例中，對應的分割槽=Math.abs("console-consumer-88985".hashCode()) % 50 = 39，即__consumer_offsets的分割槽39儲存了這個consumer group的位移資訊，下面讓我們驗證一下。
（你可以寫個Java小程式直接輸出System.out.println(Math.abs("console-consumer-88985".hashCode()) % 50);即可知道結果）

8.獲取指定consumer group的位移資訊
[[email protected] kafka_2.10-0.10.1.0]$ bin/kafka-simple-consumer-shell.sh --topic __consumer_offsets --partition 39 --broker-list h153:9092 --formatter "kafka.coordinator.GroupMetadataManager\$OffsetsMessageFormatter"
[console-consumer-88985,test,1]::[OffsetMetadata[2,NO_METADATA],CommitTime 1506433800225,ExpirationTime 1506520200225]
[console-consumer-88985,test,0]::[OffsetMetadata[2,NO_METADATA],CommitTime 1506433800225,ExpirationTime 1506520200225]
[console-consumer-88985,test,1]::[OffsetMetadata[2,NO_METADATA],CommitTime 1506433800326,ExpirationTime 1506520200326]
[console-consumer-88985,test,0]::[OffsetMetadata[2,NO_METADATA],CommitTime 1506433800326,ExpirationTime 1506520200326]
（如果將39換為其他數字則不會有上面的內容輸出）

二、Kafka 0.11客戶端管理工具AdminClient
很多使用者都有直接使用程式API操作Kafka叢集的需求。在0.11版本之前，kafka的伺服器端程式碼(即新增kafka_2.**依賴)提供了AdminClient和AdminUtils可以提供部分的叢集管理操作，但社群官網主頁並沒有給出這兩個類的使用文件。使用者只能自行檢視原始碼和測試用例才能瞭解具體的使用方法。倘若使用客戶端API的話(即新增kafka_clients依賴)，使用者必須構造特定的請求並自覺編寫程式碼向指定broker建立Socket連線併發送請求，同樣是十分繁瑣。故Kafka 0.11版本引入了客戶端的AdminClient工具。注意，雖然和原先伺服器端的AdminClient類同名，但這個工具是屬於客戶端的，因此需要在程式中新增kafka_clients依賴，比如Gradle的話則增加 compile group: 'org.apache.kafka', name: 'kafka-clients', version: '0.11.0.0'

該工具提供的所有功能包括：
1.建立topic
2.查詢所有topic
3.查詢單個topic詳情
4.刪除topic
5.修改config（包括BROKER和TOPIC資源的config）
6.查詢資源config詳情
7.建立ACL
8.查詢ACL詳情
9.刪除ACL
10.查詢整個叢集詳情

使用者使用該類的方式與Java clients的使用方式一致，不用連線Zookeeper，而是直接給定叢集中的broker列表。

下面給出一個該類的測試例項，列出了除ACL操作之外的所有操作樣例程式碼，如下所示：

import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.Config;
import org.apache.kafka.clients.admin.ConfigEntry;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.apache.kafka.clients.admin.DescribeConfigsResult;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsOptions;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.config.ConfigResource;

public class AdminClientTest {
	 
    private static final String TEST_TOPIC = "hui";
 
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "h153:9092");
 
        try (AdminClient client = AdminClient.create(props)) {
            createTopics(client);
            describeCluster(client);
            listAllTopics(client);
            describeTopics(client);
            alterConfigs(client);
            describeConfig(client);
            deleteTopics(client);
        }
    }
 
    public static void createTopics(AdminClient client) throws ExecutionException, InterruptedException {
        NewTopic newTopic = new NewTopic(TEST_TOPIC, 1, (short)1);
        CreateTopicsResult ret = client.createTopics(Arrays.asList(newTopic));
        ret.all().get();
        System.out.println("建立成功");
    }
    
    public static void describeCluster(AdminClient client) throws ExecutionException, InterruptedException {
        DescribeClusterResult ret = client.describeCluster();
        System.out.println(String.format("Cluster id-->%s ; controller-->%s", ret.clusterId().get(), ret.controller().get()));
        System.out.print("Current cluster nodes info-->");
        for (Node node : ret.nodes().get()) {
            System.out.println(node);
        }
    }
    
    public static void listAllTopics(AdminClient client) throws ExecutionException, InterruptedException {
        ListTopicsOptions options = new ListTopicsOptions();
        options.listInternal(true); // includes internal topics such as __consumer_offsets
        ListTopicsResult topics = client.listTopics(options);
        Set<String> topicNames = topics.names().get();
        System.out.println("Current topics in this cluster: " + topicNames);
    }
    
    public static void describeTopics(AdminClient client) throws ExecutionException, InterruptedException {
        DescribeTopicsResult ret = client.describeTopics(Arrays.asList(TEST_TOPIC, "__consumer_offsets"));
        Map<String, TopicDescription> topics = ret.all().get();
        for (Map.Entry<String, TopicDescription> entry : topics.entrySet()) {
            System.out.println(entry.getKey() + " ===> " + entry.getValue());
        }
    }
 
    public static void alterConfigs(AdminClient client) throws ExecutionException, InterruptedException {
        Config topicConfig = new Config(Arrays.asList(new ConfigEntry("cleanup.policy", "compact")));
        client.alterConfigs(Collections.singletonMap(
                new ConfigResource(ConfigResource.Type.TOPIC, TEST_TOPIC), topicConfig)).all().get();
    }
    
    public static void describeConfig(AdminClient client) throws ExecutionException, InterruptedException {
        DescribeConfigsResult ret = client.describeConfigs(Collections.singleton(new ConfigResource(ConfigResource.Type.TOPIC, TEST_TOPIC)));
        Map<ConfigResource, Config> configs = ret.all().get();
        for (Map.Entry<ConfigResource, Config> entry : configs.entrySet()) {
            ConfigResource key = entry.getKey();
            Config value = entry.getValue();
            System.out.println(String.format("Resource type: %s, resource name: %s", key.type(), key.name()));
            Collection<ConfigEntry> configEntries = value.entries();
            for (ConfigEntry each : configEntries) {
                System.out.println(each.name() + " = " + each.value());
            }
        }
    }

    public static void deleteTopics(AdminClient client) throws ExecutionException, InterruptedException {
        KafkaFuture<Void> futures = client.deleteTopics(Arrays.asList(TEST_TOPIC)).all();
        futures.get();
        System.out.println("刪除成功");
    }
}