应用程序重新启动时Kafka上次偏移量增加

Question

我使用的是0.10.1.1 API的高级使用者。

奇怪的是，当我关闭应用程序并重新启动它时，偏移量比上次提交的偏移量大一点，我找不到原因。

我在代码中只有一个提交点。

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());
        }
    }

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

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

    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");

Answer 1

我认为你的假设“关机前偏移量：3107169023”是基于你的关机挂钩打印的，这是正确的吗？

如果是这样的话，我看到了两个潜在的问题。

注册shutdown挂钩时，您将关闭lastCommitData字段。

当虚拟机开始其关机序列时，它将以某种未指定的顺序启动所有已注册的关机挂钩，并让它们并发运行

因此不能保证在shutdown钩子已经打印了lastCommitData值之后,使用者不会继续提交偏移量。

我建议你检查一下Kafka，看看你的应用程序关闭后，实际提交的偏移量是多少。