code: @Override
  	public Tuple2<K, V> deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
  		K key = null;
  		V value = null;
  
  		if (record.key() != null) {
  			inputDeserializer.setBuffer(record.key());
  			key = keySerializer.deserialize(inputDeserializer);
  		}
  		if (record.value() != null) {
  			inputDeserializer.setBuffer(record.value());
  			value = valueSerializer.deserialize(inputDeserializer);
  		}
  		return new Tuple2<>(key, value);
  	}
docs: ------------------------------------------------------------------------