ConsumerRecord<Object, Object> deserialize(final ProcessorContext<?, ?> processorContext,
                                               final ConsumerRecord<byte[], byte[]> rawRecord) {

        try {
            return new ConsumerRecord<>(
                rawRecord.topic(),
                rawRecord.partition(),
                rawRecord.offset(),
                rawRecord.timestamp(),
                TimestampType.CREATE_TIME,
                rawRecord.serializedKeySize(),
                rawRecord.serializedValueSize(),
                sourceNode.deserializeKey(rawRecord.topic(), rawRecord.headers(), rawRecord.key()),
                sourceNode.deserializeValue(rawRecord.topic(), rawRecord.headers(), rawRecord.value()),
                rawRecord.headers(),
                Optional.empty()
            );
        } catch (final RuntimeException deserializationException) {
            handleDeserializationFailure(deserializationExceptionHandler, processorContext, deserializationException, rawRecord, log, droppedRecordsSensor, sourceNode().name());
            return null; //  'handleDeserializationFailure' would either throw or swallow -- if we swallow we need to skip the record by returning 'null'
        }
    }
@Test
    public void shouldFailWhenDeserializationFailsAndExceptionHandlerReturnsNull() {
        try (final Metrics metrics = new Metrics()) {
            final RecordDeserializer recordDeserializer = new RecordDeserializer(
                    new TheSourceNode(
                            sourceNodeName,
                            true,
                            false,
                            "key",
                            "value"
                    ),
                    new DeserializationExceptionHandlerMock(
                            Optional.empty(),
                            rawRecord,
                            sourceNodeName,
                            taskId
                    ),
                    new LogContext(),
                    metrics.sensor("dropped-records")
            );

            final StreamsException exception = assertThrows(
                    StreamsException.class,
                    () -> recordDeserializer.deserialize(context, rawRecord)
            );
            assertEquals("Fatal user code error in deserialization error callback", exception.getMessage());
            assertInstanceOf(NullPointerException.class, exception.getCause());
            assertEquals("Invalid DeserializationExceptionHandler response.", exception.getCause().getMessage());
        }
    }