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 shouldReturnConsumerRecordWithDeserializedValueWhenNoExceptions() {
        try (final Metrics metrics = new Metrics()) {
            final RecordDeserializer recordDeserializer = new RecordDeserializer(
                    new TheSourceNode(
                            sourceNodeName,
                            false,
                            false,
                            "key",
                            "value"
                    ),
                    null,
                    new LogContext(),
                    metrics.sensor("dropped-records")
            );
            final ConsumerRecord<Object, Object> record = recordDeserializer.deserialize(null, rawRecord);
            assertEquals(rawRecord.topic(), record.topic());
            assertEquals(rawRecord.partition(), record.partition());
            assertEquals(rawRecord.offset(), record.offset());
            assertEquals("key", record.key());
            assertEquals("value", record.value());
            assertEquals(rawRecord.timestamp(), record.timestamp());
            assertEquals(TimestampType.CREATE_TIME, record.timestampType());
            assertEquals(rawRecord.headers(), record.headers());
        }
    }