apache-kafka - 高级消费者上的 commitOffsets 是否阻塞？

Question

在 Java 客户端 ( http://kafka.apache.org/documentation.html#highlevelconsumerapi ) 中，高级消费者上的 commitOffsets 会阻塞直到成功提交偏移量，还是一劳永逸？

Answer 1

高级消费者上的 commitOffsets 是否会阻塞，直到成功提交偏移量？

它看起来像commitOffsets()循环遍历每个消费者并updatePersistentPath在其偏移量发生变化时调用，如果发生变化则通过zkClient.writeData(path, getBytes(data)). 似乎在提交所有偏移量之前commitOffsets() 确实会阻塞。

这是commitOffsets()（ref）的源代码：

public void commitOffsets() {
    if (zkClient == null) {
        logger.error("zk client is null. Cannot commit offsets");
        return;
    }
    for (Entry<String, Pool<Partition, PartitionTopicInfo>> e : topicRegistry.entrySet()) {
        ZkGroupTopicDirs topicDirs = new ZkGroupTopicDirs(config.getGroupId(), e.getKey());
        for (PartitionTopicInfo info : e.getValue().values()) {
            final long lastChanged = info.getConsumedOffsetChanged().get();
            if (lastChanged == 0) {
                logger.trace("consume offset not changed");
                continue;
            }
            final long newOffset = info.getConsumedOffset();
            //path: /consumers/<group>/offsets/<topic>/<brokerid-partition>
            final String path = topicDirs.consumerOffsetDir + "/" + info.partition.getName();
            try {
                ZkUtils.updatePersistentPath(zkClient, path, "" + newOffset);
            } catch (Throwable t) {
                logger.warn("exception during commitOffsets, path=" + path + ",offset=" + newOffset, t);
            } finally {
                info.resetComsumedOffsetChanged(lastChanged);
                if (logger.isDebugEnabled()) {
                    logger.debug("Committed [" + path + "] for topic " + info);
                }
            }
        }
    }
}

对于updatePersistentPath(...)（参考）：

public static void updatePersistentPath(ZkClient zkClient, String path, String data) {
    try {
        zkClient.writeData(path, getBytes(data));
    } catch (ZkNoNodeException e) {
        createParentPath(zkClient, path);
        try {
            zkClient.createPersistent(path, getBytes(data));
        } catch (ZkNodeExistsException e2) {
            zkClient.writeData(path, getBytes(data));
        }
    }
}