如何在没有shell的Kafka 0.10.x中获取当前偏移量？

import kafka.common.TopicAndPartition;
import kafka.javaapi.OffsetResponse;
import kafka.javaapi.consumer.SimpleConsumer;
import kafka.javaapi.message.ByteBufferMessageSet;
import kafka.message.MessageAndOffset;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

public class KafkaOffsetFetcher {
    private static final int TIMEOUT = 10000;
    private static final int BUFFER_SIZE = 64 * 1024;

    public static void main(String[] args) {
        String topic = "your_topic";
        int partition = 0;
        String broker = "your_broker";
        int port = 9092;

        long offset = getLatestOffset(topic, partition, broker, port);
        System.out.println("Latest offset: " + offset);
    }

    private static long getLatestOffset(String topic, int partition, String broker, int port) {
        SimpleConsumer consumer = new SimpleConsumer(broker, port, TIMEOUT, BUFFER_SIZE, "offsetLookup");
        TopicAndPartition topicAndPartition = new TopicAndPartition(topic, partition);
        Map<TopicAndPartition, Long> requestInfo = new HashMap<>();
        requestInfo.put(topicAndPartition, kafka.api.OffsetRequest.LatestTime());
        kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion(), consumer.clientId());

        OffsetResponse response = consumer.getOffsetsBefore(request);
        if (response.hasError()) {
            System.out.println("Error fetching data Offset Data the Broker. Reason: " + response.errorCode(topic, partition));
            return -1;
        }

        long[] offsets = response.offsets(topic, partition);
        consumer.close();

        if (offsets.length > 0) {
            return offsets[0];
        } else {
            return -1;
        }
    }
}