You could write code like this to fetch the offsets for a group:
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();
}
}
private byte[] issueRequestAndWaitForResponse(Socket socket, byte[] request) throws IOException {
sendRequest(socket, request);
return getResponse(socket);
}
private void sendRequest(Socket socket, byte[] request) throws IOException {
DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
dos.writeInt(request.length);
dos.write(request);
dos.flush();
}
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();
}
}
}
private Socket connect(String hostName, int port) throws IOException {
return new Socket(hostName, port);
}
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;
}
// Get offsets of a given topic for a group
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);
}
// Get offsets of all topics for a group
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();
}
AdminClient.createSimplePlaintext("localhost:9092").listGroupOffsets
– amethystic