class ProducerRecord<K, V> {
   K key;
   V value;
   Map<Integer, byte[]> headers;
}

class ConsumerRecord<K, V> {
	K key;
	V value;
	Map<Integer, byte[]> headers;
}

interface Serializer<T> {
	byte[] serialize(String topic, T data, Map<Integer, byte[]> headers);
}

interface Deserializer<T> {
	T deserialize(String topic, byte[] data, Map<Integer, byte[]> headers);
}

interface ProducerPostSerializationInterceptor {
	//key and value have already been serialized, but headers not
	ProducerRecord<byte[], byte[]> onSend(ProducerRecord<byte[], byte[]> serializedRecord);
}

AvroSerializer<T extends GenericRecord> implements Serializer<T> Deserializer<T>{
	@Override
	byte[] serialize(String topic, T data, Map<Integer, byte[]> headers) {
		byte[] schemaMd5 = schemaRegistryClient.depositSchemaIfNew(...)
		headers.put(CONTENT_TYPE, "application/avro-binary".getBytes());
		headers.put(AVRO_SCHEMA_MD5, schemaMd5);
        return doSerialize(data);
	}

	@Override
	T deserialize(String topic, byte[] data, Map<Integer, byte[]> headers) {
		byte[] schemaMd5 = headers.get(AVRO_SCHEMA_MD5);
        return deDeserialize(schemaRegistryClient.get(schemaMd5), data);
	}
 }

PolyglotDeserializer implements Deserializer<T> {
	Map<String, Deserializer> byContentType = ...

	@Override
	T deserialize(String topic, byte[] data, Map<Integer, byte[]> headers) {
		String type = new String(headers.get(CONTENT_TYPE));
        return byContentType.get(type).deserialize(...);
	}
}

class Entry implements Serializable {
   String byWhom;
   String producedIntoCluster;
   String producedIntoTopic;
   long when;
   String readFromCluster;
   String readFromTopic;
}

PaperTrailInterceptor implements ProducerInterceptor<K, V>, ConsumerInterceptor<K, V> {
    @Override
	ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
		List<Entry> trail = getOrCreateTrail(record.headers, false); //if we're on a mirrormaker the consumer has already set the latest entry
		Entry latest = trail.get(trail.size()-1);
		latest.setProducedIntoCluster(...);
        latest.setProducedIntoTopic(record.topic);
		record.headers.put(PAPER_TRAIL, serialize(trail));
	}

	@Override
	ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records) {
		for each record {
			List<Entry> trail = getOrCreateTrail(record.headers, true);
			Entry latest = trail.get(trail.size()-1);
    	    latest.setReadFromCluster(...);
    	    latest.setReadFromTopic(record.topic);
			record.headers.put(PAPER_TRAIL, serialize(trail));
		}
	}

	private Entry getOrCreateTrail(Map<Integer, byte[]> headers, boolean createRegardless) {
		List<Entry> trail = deserialize(headers.get(PAPER_TRAIL));
		Entry latest;
		if (trail == null || createRegardless) {
			trail = new ArrayList<>(1);            
            latest = new Entry(hostName, now);
            trail.add(latest);
		}
		return trail;
	}
}

TraceInterceptor implements ProducerInterceptor<K, V>, ConsumerInterceptor<K, V> {

	@Override
	ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
		Object traceCookie = SomeThreadLocalContext.getContext();
		if (traceCookie != null) { //this produce call is part of some larger tx, pass it on
			record.headers.put(TRACE_COOKIE, serialize(traceCookie));
		}
	}

	@Override
	ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records) {
		//this would really benefit from a per-record onConsume()...
	}
}

Signer implements ProducerPostSerializationInterceptor {
	@Override
	void onSend(ProducerRecord<byte[], byte[]> record) {
		byte[] signature = sign(record.value);
		record.headers.put(SIGNATURE, signature);
	}
}

SmartCompressor implements ProducerPostSerializationInterceptor {
	@Override
	void onSend(ProducerRecord<byte[], byte[]> record) {
		//could be made much smarter, by CONTENT_TYPE of even sniffing for known payload types
		byte[] compressed = compress(record.value);
		if (compressed.length < record.value.length) {
			//only if content is actually compressible
			record.value = compressed;
			record.headers.put(COMPRESSION, "gzip".getBytes());
		}
	}
}

Slicer implements ProducerPostSerializationInterceptor {
	@override
	List<ProducerRecord<byte[], byte[]>> onSend(ProducerRecord<byte[], byte[]> serializedRecord) { //API would need to support returning List<Record>
		if (sizeOf(serializedRecord) < maxSingleRecordSize) {
			return Collections.singletonList(serializedRecord); //in other words, dont touch it
		}
		byte[][] chunks = slice(serializedRecord.value, maxSingleRecordSize - sizeOfOurExtraHeader);
		int numChunks = chunks.length;
		List<ProducerRecord<byte[], byte[]>> chunkRecords = new ArrayList<>(numChunks);
		for (int i=0; i<numChunks; i++) {
			byte[] chunk = chunks[i];
			ProducerRecord<byte[], byte[]> chunkRecord;
			if (i==0) {
				//1st chunk retains all headers
				chunkRecord = serializedRecord;
			} else {
				chunkRecord = new ProducerRecord(serializedRecord.key, chunk, new HashMap<>(1)); //empty header map for chunks 2+
			}
			chunkRecord.headers.put(CHUNK, (new ChunkDescriptor(i, numChunks)).serialize());
			chunkRecords.add(chunkRecord);
		}
		return chunkRecords;
	}
}

Glue implements ConsumerInterceptor<K, V> {
	List<ConsumerRecord<K, V>> chunksSoFar = new ArrayList<>(); //making assumptions about consumer interceptor threading guarantees ...

	@Override
	ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records) {
		ConsumerRecords<K, V> reassembled = new ...;

		for (ConsumerRecord<K, V> record : records) {
			ChunkDescriptor desc = getOrNull(record.headers);
			if (desc == null) {
				//this record is not part of a bigger record, so "pass" as-is
				reassembled.add(record);
				continue;
			}
			chunksSoFar.add(record);
			if (desc.getNum() == desc.getCount()) { //over simplified
				reassembled.add(pieceBack(chunksSoFar));
				chunksSoFar.clear();
			}
		}
		
		return reassembled;
	}
}