@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()
        );
        updateProcessorContext(node, time.milliseconds(), recordContext);

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

        try {
            maybeMeasureLatency(() -> punctuator.punctuate(timestamp), time, punctuateLatencySensor);
        } catch (final TimeoutException timeoutException) {
            if (!eosEnabled) {
                throw timeoutException;
            } else {
                record = null;
                throw new TaskCorruptedException(Collections.singleton(id));
            }
        } catch (final FailedProcessingException e) {
            throw createStreamsException(node.name(), e.getCause());
        } catch (final TaskCorruptedException | TaskMigratedException e) {
            throw e;
        } catch (final RuntimeException processingException) {
            final ErrorHandlerContext errorHandlerContext = new DefaultErrorHandlerContext(
                null,
                recordContext.topic(),
                recordContext.partition(),
                recordContext.offset(),
                recordContext.headers(),
                node.name(),
                id()
            );

            final ProcessingExceptionHandler.ProcessingHandlerResponse response;
            try {
                response = Objects.requireNonNull(
                    processingExceptionHandler.handle(errorHandlerContext, null, processingException),
                    "Invalid ProcessingExceptionHandler response."
                );
            } catch (final RuntimeException fatalUserException) {
                log.error(
                    "Processing error callback failed after processing error for record: {}",
                    errorHandlerContext,
                    processingException
                );
                throw new FailedProcessingException("Fatal user code error in processing error callback", fatalUserException);
            }

            if (response == ProcessingExceptionHandler.ProcessingHandlerResponse.FAIL) {
                log.error("Processing exception handler is set to fail upon" +
                        " a processing error. If you would rather have the streaming pipeline" +
                        " continue after a processing error, please set the " +
                        PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG + " appropriately.");

                throw createStreamsException(node.name(), processingException);
            } else {
                droppedRecordsSensor.record();
            }
        } finally {
            processorContext.setCurrentNode(null);
        }
    }
@Test
    public void punctuateShouldThrowFailedProcessingExceptionWhenProcessingExceptionHandlerThrowsAnException() {
        when(stateManager.taskId()).thenReturn(taskId);
        when(stateManager.taskType()).thenReturn(TaskType.ACTIVE);
        task = createStatelessTask(createConfig(
            AT_LEAST_ONCE,
            "100",
            LogAndFailExceptionHandler.class.getName(),
            CrashingProcessingExceptionHandler.class.getName()
        ));

        final FailedProcessingException streamsException = assertThrows(
            FailedProcessingException.class,
            () -> task.punctuate(processorStreamTime, 1, PunctuationType.STREAM_TIME, timestamp -> {
                throw new KafkaException("KABOOM!");
            })
        );

        assertEquals("Fatal user code error in processing error callback", streamsException.getMessage());
        assertEquals("KABOOM from ProcessingExceptionHandlerMock!", streamsException.getCause().getMessage());
    }