class SourceNodeRecordDeserializer implements RecordDeserializer {

    @SuppressWarnings("deprecation")
    @Override
    public ConsumerRecord<Object, Object> deserialize(final ConsumerRecord<byte[], byte[]> rawRecord) {
        final Object key;
        try {
            key = sourceNode.deserializeKey(rawRecord.topic(), rawRecord.headers(), rawRecord.key());
        } catch (Exception e) {
            throw new StreamsException(format("Failed to deserialize key for record. topic=%s, partition=%d, offset=%d",
                                              rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e);
        }

        final Object value;
        try {
            value = sourceNode.deserializeValue(rawRecord.topic(), rawRecord.headers(), rawRecord.value());
        } catch (Exception e) {
            throw new StreamsException(format("Failed to deserialize value for record. topic=%s, partition=%d, offset=%d",
                                              rawRecord.topic(), rawRecord.partition(), rawRecord.offset()), e);
        }

        return new ConsumerRecord<>(rawRecord.topic(), rawRecord.partition(), rawRecord.offset(),
                                    rawRecord.timestamp(), TimestampType.CREATE_TIME,
                                    rawRecord.checksum(),
                                    rawRecord.serializedKeySize(),
                                    rawRecord.serializedValueSize(), key, value);

    }

      SourceNodeRecordDeserializer(final SourceNode sourceNode,
                                 final DeserializationExceptionHandler deserializationExceptionHandler);

}