Cancelling Scheduled Punctuator after a single iteration

[viktorjucas]

Is there an easy way to cancel a scheduled punctuator after the first time it runs?

[guozhangwang]

@viktorjucas Yes, note that ProcessorContex#schedule returns a Cancellable event. You can cancel it even within the punctuate call. For example:

Cancellable scheduled;
scheduled = context.schedule(... timestamp -> scheduled.cancel);

[viktorjucas]

I have tried that and it doesn't seem to be cancelling.

public class Transformer implements Transformer<> {
    private ProcessorContext context;
    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();
    }
}

[guozhangwang]

Which version are you using? Note that this functionality is fixed in 1.1.1: https://issues.apache.org/jira/browse/KAFKA-6748

[viktorjucas]

That's the reason. I was using 1.1.0.

[viktorjucas]

Closed prematurely.

Is v1.1.1 available through Maven already?

[guozhangwang]

No it is not released out yet: https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams

If you have to cancel the even within the punctuate call, you can consider backporting that fix into your source code build.

[mjsax]

@viktorjucas Kafka 1.1.1 is released now. Can you try it out and verify the fix so we can close this issue?