title: Cancel Punctuator on Kafka Streams after is triggered
body: <p>I create a scheduled punctuator on a transformer and I schedule it to run on a periodical basis (using kafka v2.1.0). Every time I accept a specific key I do create a new one like this</p>
  
  <pre><code>        scheduled = context.schedule (Duration.ofMillis(scheduleTime),
               PunctuationType.WALL_CLOCK_TIME,new CustomPunctuator(context, customStateStoreName));
  </code></pre>
  
  <p>My issue is that all these punctuators I create run constantly and I cannot find a way to cancel them. I found a snippet in the internet to use</p>
  
  <pre><code>    private Cancellable scheduled;
  
      @Override
      public void init(PorcessorContext processContext) {
          this.context = processorContext;
          scheduled = context.schedule(TimeUnit.SECONDS.toMillis(5), PunctuationType.WALL_CLOCK_TIME,
                                       this::punctuateCancel);
      }
  
      private void punctuateCancel(long timestamp) {
          scheduled.cancel();
      }
  </code></pre>
  
  <p>but this unfortunately seems to cancel only the latest created Punctuator.</p>
  
  <p>I am editing my post just to give some further insight regarding my approach and how this is related with comments made by wardzinia. So my approach is pretty similar just uses a Map because I need to have only one punctuator active per event key so in my Transformer class I initiate</p>
  
  <pre><code>   private Map&lt;String,Cancellable&gt; scheduled  = new HashMap&lt;&gt;();
  </code></pre>
  
  <p>And on my transform method I do execute the code below</p>
  
  <pre><code>{
   final Cancellable cancelSched = scheduled.get(recordKey);
  
   // Every time I get a new event I cancel my previous Punctuator
   // and schedule a new one ( context.schedule a few lines later)
  
   if(cancelSched != null)
      cancelSched.cancel();
  
   // This is supposed to work like a closure by capturing the currentCancellable which in the next statement
   // is moved to the scheduled map. Scheduled map at any point will have the only active Punctuator for a
   // specific String as it is constantly renewed
   // Note: Previous registered punctuators have already been cancelled as it can be seen by the previous
   // statement (cancelSched.cancel();)
  
   Cancellable currentCancellable = context.schedule(Duration.ofMillis(scheduleTime), PunctuationType.WALL_CLOCK_TIME,
                  new CustomPunctuator(context, recordKey ,()-&gt; scheduled ));
  
   // Update Active Punctuators for a specific key.
  
   scheduled.put(recordKey,currentCancellable);   
  }
  </code></pre>
  
  <p>And I use that registered callback on my Punctuator punctuate method to cancel the last active Punctuator
  after it has started. It seems to work (not sure though) but it feels very "hacky" and not the kind of solution 
  that it is certainly desirable.</p>
  
  <p>So how can I cancel a punctuator after is triggered. Is there a way to cope with this issue ?</p>