class RecordDeserializer {

    @SuppressWarnings("deprecation")
    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.checksum(),
                rawRecord.serializedKeySize(),
                rawRecord.serializedValueSize(),
                sourceNode.deserializeKey(rawRecord.topic(), rawRecord.headers(), rawRecord.key()),
                sourceNode.deserializeValue(rawRecord.topic(), rawRecord.headers(), rawRecord.value()),
                rawRecord.headers()
            );
        } catch (final Exception deserializationException) {
            final DeserializationExceptionHandler.DeserializationHandlerResponse response;
            try {
                response = deserializationExceptionHandler.handle(processorContext, rawRecord, deserializationException);
            } catch (final Exception fatalUserException) {
                log.error(
                    "Deserialization error callback failed after deserialization error for record {}",
                    rawRecord,
                    deserializationException);
                throw new StreamsException("Fatal user code error in deserialization error callback", fatalUserException);
            }

            if (response == DeserializationExceptionHandler.DeserializationHandlerResponse.FAIL) {
                throw new StreamsException("Deserialization exception handler is set to fail upon" +
                    " a deserialization error. If you would rather have the streaming pipeline" +
                    " continue after a deserialization error, please set the " +
                    DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG + " appropriately.",
                    deserializationException);
            } else {
                log.warn(
                    "Skipping record due to deserialization error. topic=[{}] partition=[{}] offset=[{}]",
                    rawRecord.topic(),
                    rawRecord.partition(),
                    rawRecord.offset(),
                    deserializationException
                );
                droppedRecordsSensor.record();
                return null;
            }
        }
    }

      RecordDeserializer(final SourceNode<?, ?, ?, ?> sourceNode,
                       final DeserializationExceptionHandler deserializationExceptionHandler,
                       final LogContext logContext,
                       final Sensor droppedRecordsSensor);

}