<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka-clients</artifactId>
  <version>0.10.2.0</version>
</dependency>

compile group: 'org.apache.kafka', name: 'kafka-clients', version: '0.10.2.0'

/**
   * 發(fā)送請(qǐng)求主方法
   * @param host     目標(biāo)broker的主機(jī)名
   * @param port     目標(biāo)broker的端口
   * @param request    請(qǐng)求對(duì)象
   * @param apiKey    請(qǐng)求類型
   * @return       序列化后的response
   * @throws IOException
   */
  public ByteBuffer send(String host, int port, AbstractRequest request, ApiKeys apiKey) throws IOException {
    Socket socket = connect(host, port);
    try {
      return send(request, apiKey, socket);
    } finally {
      socket.close();
    }
  }

  /**
   * 發(fā)送序列化請(qǐng)求并等待response返回
   * @param socket      連向目標(biāo)broker的socket
   * @param request      序列化后的請(qǐng)求
   * @return         序列化后的response
   * @throws IOException
   */
  private byte[] issueRequestAndWaitForResponse(Socket socket, byte[] request) throws IOException {
    sendRequest(socket, request);
    return getResponse(socket);
  }

  /**
   * 發(fā)送序列化請(qǐng)求給socket
   * @param socket      連向目標(biāo)broker的socket
   * @param request      序列化后的請(qǐng)求
   * @throws IOException
   */
  private void sendRequest(Socket socket, byte[] request) throws IOException {
    DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
    dos.writeInt(request.length);
    dos.write(request);
    dos.flush();
  }

  /**
   * 從給定socket處獲取response
   * @param socket      連向目標(biāo)broker的socket
   * @return         獲取到的序列化后的response
   * @throws IOException
   */
  private byte[] getResponse(Socket socket) throws IOException {
    DataInputStream dis = null;
    try {
      dis = new DataInputStream(socket.getInputStream());
      byte[] response = new byte[dis.readInt()];
      dis.readFully(response);
      return response;
    } finally {
      if (dis != null) {
        dis.close();
      }
    }
  }

  /**
   * 創(chuàng)建Socket連接
   * @param hostName     目標(biāo)broker主機(jī)名
   * @param port       目標(biāo)broker服務(wù)端口, 比如9092
   * @return         創(chuàng)建的Socket連接
   * @throws IOException
   */
  private Socket connect(String hostName, int port) throws IOException {
    return new Socket(hostName, port);
  }

  /**
   * 向給定socket發(fā)送請(qǐng)求
   * @param request    請(qǐng)求對(duì)象
   * @param apiKey    請(qǐng)求類型, 即屬于哪種請(qǐng)求
   * @param socket    連向目標(biāo)broker的socket
   * @return       序列化后的response
   * @throws IOException
   */
  private ByteBuffer send(AbstractRequest request, ApiKeys apiKey, Socket socket) throws IOException {
    RequestHeader header = new RequestHeader(apiKey.id, request.version(), "client-id", 0);
    ByteBuffer buffer = ByteBuffer.allocate(header.sizeOf() + request.sizeOf());
    header.writeTo(buffer);
    request.writeTo(buffer);
    byte[] serializedRequest = buffer.array();
    byte[] response = issueRequestAndWaitForResponse(socket, serializedRequest);
    ByteBuffer responseBuffer = ByteBuffer.wrap(response);
    ResponseHeader.parse(responseBuffer);
    return responseBuffer;
  }

/**
   * 創(chuàng)建topic
   * 由于只是樣例代碼，有些東西就硬編碼寫到程序里面了(比如主機(jī)名和端口)，各位看官自行修改即可
   * @param topicName       topic名
   * @param partitions      分區(qū)數(shù)
   * @param replicationFactor   副本數(shù)
   * @throws IOException
   */
  public void createTopics(String topicName, int partitions, short replicationFactor) throws IOException {
    Map<String, CreateTopicsRequest.TopicDetails> topics = new HashMap<>();
    // 插入多個(gè)元素便可同時(shí)創(chuàng)建多個(gè)topic
    topics.put(topicName, new CreateTopicsRequest.TopicDetails(partitions, replicationFactor));
    int creationTimeoutMs = 60000;
    CreateTopicsRequest request = new CreateTopicsRequest.Builder(topics, creationTimeoutMs).build();
    ByteBuffer response = send("localhost", 9092, request, ApiKeys.CREATE_TOPICS);
    CreateTopicsResponse.parse(response, request.version());
  }

/**
   * 獲取某個(gè)consumer group下的某個(gè)topic分區(qū)的位移
   * @param groupID      group id
   * @param topic       topic名
   * @param parititon     分區(qū)號(hào)
   * @throws IOException
   */
  public void getOffsetForPartition(String groupID, String topic, int parititon) throws IOException {
    TopicPartition tp = new TopicPartition(topic, parititon);
    OffsetFetchRequest request = new OffsetFetchRequest.Builder(groupID, singletonList(tp))
        .setVersion((short)2).build();
    ByteBuffer response = send("localhost", 9092, request, ApiKeys.OFFSET_FETCH);
    OffsetFetchResponse resp = OffsetFetchResponse.parse(response, request.version());
    OffsetFetchResponse.PartitionData partitionData = resp.responseData().get(tp);
    System.out.println(partitionData.offset);
  }

/**
   * 獲取某個(gè)consumer group下所有topic分區(qū)的位移信息
   * @param groupID      group id
   * @return         (topic分區(qū) --> 分區(qū)信息)的map
   * @throws IOException
   */
  public Map<TopicPartition, OffsetFetchResponse.PartitionData> getAllOffsetsForGroup(String groupID) throws IOException {
    OffsetFetchRequest request = new OffsetFetchRequest.Builder(groupID, null).setVersion((short)2).build();
    ByteBuffer response = send("localhost", 9092, request, ApiKeys.OFFSET_FETCH);
    OffsetFetchResponse resp = OffsetFetchResponse.parse(response, request.version());
    return resp.responseData();
  }