@SuppressWarnings("unchecked")
    @Override
    public void punctuate(final ProcessorNode<?, ?, ?, ?> node,
                          final long timestamp,
                          final PunctuationType type,
                          final Punctuator punctuator) {
        if (processorContext.currentNode() != null) {
            throw new IllegalStateException(String.format("%sCurrent node is not null", logPrefix));
        }

        // when punctuating, we need to preserve the timestamp (this can be either system time or event time)
        // while other record context are set as dummy: null topic, -1 partition, -1 offset and empty header
        final ProcessorRecordContext recordContext = new ProcessorRecordContext(
            timestamp,
            -1L,
            -1,
            null,
            new RecordHeaders(),
            null
        );
        updateProcessorContext(node, time.milliseconds(), recordContext);

        if (log.isTraceEnabled()) {
            log.trace("Punctuating processor {} with timestamp {} and punctuation type {}", node.name(), timestamp, type);
        }

        try {
            maybeMeasureLatency(() -> node.punctuate(timestamp, punctuator), time, punctuateLatencySensor);
        } catch (final StreamsException e) {
            throw e;
        } catch (final RuntimeException e) {
            throw new StreamsException(String.format("%sException caught while punctuating processor '%s'", logPrefix, node.name()), e);
        } finally {
            processorContext.setCurrentNode(null);
        }
    }
@throws IllegalStateException if the current node is not null
@throws TaskMigratedException if the task producer got fenced (EOS only)
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java