@@ -0,0 +1,1033 @@
+package reactor.kafka.support;%0A%0Aimport org.apache.kafka.common.serialization.Serializer;%0Aimport org.apache.kafka.common.serialization.StringSerializer;%0Aimport reactor.pipe.key.Key;%0A%0Aimport java.util.Arrays;%0Aimport java.util.Map;%0Aimport java.util.stream.Collectors;%0A%0Apublic class KeySerializer implements Serializer%3CKey%3E %7B%0A%0A  private final StringSerializer stringSerializer;%0A%0A  public KeySerializer() %7B%0A    this.stringSerializer = new StringSerializer();%0A  %7D%0A%0A  @Override%0A  public void configure(Map%3CString, ?%3E map, boolean b) %7B%0A    stringSerializer.configure(map, b);%0A  %7D%0A%0A  @Override%0A  public byte%5B%5D serialize(String topic, Key key) %7B%0A    return stringSerializer.serialize(topic,%0A                                      String.join(%22,%22,%0A                                                  Arrays.stream(key.getObjects())%0A                                                        .map(Object::toString)%0A                                                        .collect(Collectors.toList())));%0A  %7D%0A%0A  @Override%0A  public void close() %7B%0A  %7D%0A%7D%0A

Add a serializer for key