title: Kafka Streams low level processor API's punctuate don't run periodically
body: <p>I have one <code>processor</code> named <code>AddCashProcessor</code>.</p>
  
  <p>In <code>AddCashProcessor</code>, i maintain a <code>KeyValueStore&lt;String, HashSet&lt;String&gt;&gt;</code> to record the userid of who pay for the item by <code>process</code> method. The code like this:</p>
  
  <pre><code>@Override
  public void process(String key, String value) {
      HashSet&lt;String&gt; set = Optional.ofNullable(store.get(key)).orElse(new HashSet&lt;&gt;());
      set.add(value);
      store.put(key, set);
  }
  </code></pre>
  
  <p>And in <code>punctuate</code> method of a class named <code>AddCashPunctuator</code> which <code>implements the Punctuator interface</code>, i get the <code>HashSet.size()</code> to insert it to MySQL:</p>
  
  <pre><code>@Override
  public void punctuate(long l) {
      List&lt;String&gt; updateSqls = new ArrayList&lt;&gt;();
  
      KeyValueIterator&lt;String, HashSet&lt;String&gt;&gt; iter = store.all();
      while (iter.hasNext()) {
          KeyValue&lt;String, HashSet&lt;String&gt;&gt; entry = iter.next();
              int size = entry.getValue().size();
              ....
          }
      }
      iter.close();
  
      MySqlUtils.update(updateSqls);
  
  }
  </code></pre>
  
  <p>The <code>AddCashPunctuator</code> is registered in <code>init</code> method in <code>AddCashProcessor</code>, like this:</p>
  
  <pre><code>@Override
  public void init(ProcessorContext context) {
      this.context = context;
      ....
      this.context.schedule(30000L, PunctuationType.WALL_CLOCK_TIME, new AddCashPunctuator());
  }
  </code></pre>
  
  <p>I want to execute the <code>punctuate</code> method every 30 seconds. But it not do it like this.Sometimes it runs well, sometimes pause. When the end of the pause, many times are performed.</p>
  
  <p>Why is it this? Is it because the data of the <code>HashSet</code> or <code>KeyValueStore</code> too big?
  My kafka streams version is 1.0.0.  And my kafka version is 0.10.1.1.</p>
  
  <p>Thanks!</p>