Status

Current state: Under Discussion [One of "Under Discussion", "Accepted", "Rejected"] Accepted (vote thread)

Discussion thread: here

JIRA: KAFKA-6923

...

public interface Serializer<T> extends Closeable {

    void configure(Map<String, ?> configs, boolean isKey);

    default byte[] serialize(String topic, T data) {
        return new byte[0];
    }

    default byte[] serialize(String topic, Headers headers, T data) {                        // This is the new method
        return serialize(topic, data);
    }

    @Override
    void close();
}

...

public interface Deserializer<T> extends Closeable {

    void configure(Map<String, ?> configs, boolean isKey);

    default T deserialize(String topic, byte[] data) {
        return null;
    }

    default T deserialize(String topic, Headers headers, byte[] data) {                      // This is the new method
        return deserialize(topic, data);
    }

    @Override
    void close();
}

...

Review existing unit tests and system tests.

Manual verification of binary compatibility: existing ExtendedSerializer implementations should work without recompiling the application.

Rejected Alternatives

The code examples here are made with the Deserializer class but they are completely valid for the Serializer as well.

Default Implementations for the "headerless" methods

There are options to provide a default implementation: return a dummy value or throw an exception as shown below. This means that we'll have an interface where there are no methods enforced. Furthermore we should somehow suggest users that which method should they implement. To achieve this we could deprecate the 2 parameter ("headerless") method but this would litter the code base with warnings as we're still using this method in a lot of places (like the serializers and deserializers).

public interface Serializer<T> extends Closeable {

    void configure(Map<String, ?> configs, boolean isKey);
    
	@Deprecated
    default byte[] serialize(String topic, T data) {
        return new byte[0];
		// throw new UnsupportedOperationException("Method not implemented");
    }

    default byte[] serialize(String topic, Headers headers, T data) {                        // This is the new method
        return serialize(topic, data);
    }
    @Override
    void close();
}

public interface Deserializer<T> extends Closeable {

    void configure(Map<String, ?> configs, boolean isKey);

	@Deprecated
    default T deserialize(String topic, byte[] data) {
        return null;
		// throw new UnsupportedOperationException("Method not implemented");
    }

    default T deserialize(String topic, Headers headers, byte[] data) {                      // This is the new method
        return deserialize(topic, data);
    }

    @Override
    void close();
}

Propagate to more complicated

...