[jira] [Created] (KAFKA-12346) punctuate is called at twice the duration passed as the first argument to Processor.Schedule (with PunctuationType.WALL_CLOCK_TIME)

["Arindam Ray (Jira)" <ji...@apache.org>]

Arindam Ray created KAFKA-12346:
-----------------------------------

             Summary: punctuate is called at twice the duration passed as the first argument to Processor.Schedule (with PunctuationType.WALL_CLOCK_TIME)
                 Key: KAFKA-12346
                 URL: https://issues.apache.org/jira/browse/KAFKA-12346
             Project: Kafka
          Issue Type: Bug
          Components: streams
    Affects Versions: 2.7.0
            Reporter: Arindam Ray


A stream transform called with the idiom below causes punctuate to be called at twice the duration of the argument passed
{code:java}
        .transform(new TransformerSupplier[String, TimeStampedString, KeyValue[String, TimeStampedString]]() {
          override def get(): Transformer[String, TimeStampedString, KeyValue[String, TimeStampedString]] = new Transformer[String, TimeStampedString, KeyValue[String, TimeStampedString]] {            override def init(context: ProcessorContext): Unit = {
              val store = context.getStateStore(stateStoreName).asInstanceOf[KeyValueStore[String, ValueAndTimestamp[TimeStampedString]]]
              context.schedule(scanFrequency,
                PunctuationType.WALL_CLOCK_TIME,
                new Punctuator {
                  override def punctuate(timestamp: Long): Unit = {
                    logger.info(s"Punctuate invoked with timestamp : ${Instant.ofEpochMilli(timestamp)}")
                  }
                }
              )
            }            override def transform(key: String, value: TimeStampedString): KeyValue[String, TimeStampedString] = {
              // no need to return anything here, the Punctuator will emit the records when necessary
              null
            }            override def close(): Unit = {}
          }
        },          /**
           * register that this Transformer needs to be connected to our state store.
           */
          stateStoreName
        )
{code}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)