我有一个 processor 命名 AddCashProcessor .
在 AddCashProcessor ，我保持 KeyValueStore<String, HashSet<String>> 记录付款人的用户标识 process 方法。代码如下：

@Override
public void process(String key, String value) {
    HashSet<String> set = Optional.ofNullable(store.get(key)).orElse(new HashSet<>());
    set.add(value);
    store.put(key, set);
}

而且在 punctuate 名为的类的方法 AddCashPunctuator 哪个 implements the Punctuator interface ，我明白了 HashSet.size() 要将其插入mysql：

@Override
public void punctuate(long l) {
    List<String> updateSqls = new ArrayList<>();

    KeyValueIterator<String, HashSet<String>> iter = store.all();
    while (iter.hasNext()) {
        KeyValue<String, HashSet<String>> entry = iter.next();
            int size = entry.getValue().size();
            ....
        }
    }
    iter.close();

    MySqlUtils.update(updateSqls);

}

这个 AddCashPunctuator 注册于 init 中的方法 AddCashProcessor ，如下所示：

@Override
public void init(ProcessorContext context) {
    this.context = context;
    ....
    this.context.schedule(30000L, PunctuationType.WALL_CLOCK_TIME, new AddCashPunctuator());
}

我要执行 punctuate 方法每30秒。但它不是这样做的，有时运行良好，有时停顿。暂停结束时，执行多次。
为什么是这个？是因为 HashSet 或者 KeyValueStore 太大了？我的Kafka流版本是1.0.0。我的Kafka版本是0.10.1.1。
谢谢！