Kafka 1.0.1中对Processor.close()的 JavaDoc描述说:
注意:不要关闭任何流托管资源,例如此处的StateStores,因为它们是由库管理的。
此外,StateStore.close()的JavaDoc描述说:
用户只需要实现此功能,而永远不需要显式调用此api,因为在必要时库会自动调用它
但是我在文档中找到了一个示例,其中状态存储在该方法内被显式关闭:
@Override public void close() { // close the key-value store kvStore.close(); }
所以,我有点迷路了。我应该关闭内的状态存储Processor
吗?
不,您不应该关闭商店。
示例代码段是错误的。感谢您指出。我打开了一个公关来修复它:github.com/apache/kafka/pull/4667