我有一个javaakka应用程序，它从kafka读取、处理消息并手动提交。
我使用的是0.10.1.1api的高级使用者。
奇怪的是，当我关闭应用程序并再次启动它时，偏移量比上次提交的偏移量大一点，我找不到原因。
代码中只有一个提交点。

else if(message.getClass() == ProcessedBatches.class) {
        try {
            Logger.getRootLogger().info("[" + this.name + "/Reader] Commiting ...");
            ProcessedBatches msg = (ProcessedBatches) message;
            consumer.commitSync(msg.getCommitInfo());
            lastCommitData = msg.getCommitInfo();
            lastCommit = System.currentTimeMillis();
        } catch (CommitFailedException e) {
            Logger.getRootLogger().info("[" + this.name + "/Reader] Failed to commit... Last commit: " + lastCommit + " | Last batch: " + lastBatch + ". Current uncommited messages: " + uncommitedMessages);
            self().tell(HarakiriMessage.getInstance(), self());
        }
    }

提交之后，我将偏移量hashmap保存在lastcommitdata中，以便对其进行调试。
然后我添加了一个shutdown钩子来打印lastcommitdata变量，以检查每个分区提交的最后一个偏移量是多少。

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
        String output = 
                "############## SHUTTING DOWN CONSUMER ############### \n" + 
                lastCommitData+"\n";
        System.out.println(output);
    }));

我还有一个consumerrebalancelistener，在consumer启动时检查每个分区的开始位置。

new ConsumerRebalanceListener() {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> collection) {}

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> collection) {
            for (TopicPartition p:collection
                 ) {
                System.out.println("Starting position "+p.toString()+":" + consumer.position(p));
            }
            coordinator.setRebalanceTimestamp(System.currentTimeMillis());
        }
    });

一个分区的示例：
停机前偏移量：3107169023
分配分区时的偏移量：3107180350
正如您所看到的，每封邮件之间几乎有10万条消息。
消费者属性如下：

Properties props = new Properties();
    props.put("bootstrap.servers", bootstrapServers);
    props.put("group.id", group_id);
    props.put("enable.auto.commit", "false");
    props.put("auto.commit.interval.ms", "100000000");
    props.put("session.timeout.ms", "10000");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
    props.put("max.poll.records", "40000");
    props.put("auto.offset.reset", "latest");

我不确定我做错了什么。