[KAFKA-7273] Converters should have access to headers.

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 2.4.0
• Component/s: connect
• Assignee: Jeremy Custenborder
• Reporter: Jeremy Custenborder
• Reviewer: Randall Hauch
• Created: 09/Aug/18 21:25
• Updated: 22/Jan/20 16:46
• Resolved: 25/Sep/19 17:02

Description

I found myself wanting to build a converter that stored additional type information within headers. The converter interface does not allow a developer to access to the headers in a Converter. I'm not suggesting that we change the method for serializing them, rather that org.apache.kafka.connect.header.Headers be passed in for fromConnectData and toConnectData. For example something like this.

import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaAndValue;
import org.apache.kafka.connect.header.Headers;
import org.apache.kafka.connect.storage.Converter;

public interface Converter {
  default byte[] fromConnectData(String topic, Headers headers, Schema schema, Object object) {
    return fromConnectData(topic, schema, object);
  }

  default SchemaAndValue toConnectData(String topic, Headers headers, byte[] payload) {
    return toConnectData(topic, payload);
  }

  void configure(Map<String, ?> var1, boolean var2);

  byte[] fromConnectData(String var1, Schema var2, Object var3);

  SchemaAndValue toConnectData(String var1, byte[] var2);
}

This would be a similar approach to what was already done with ExtendedDeserializer and ExtendedSerializer in the Kafka client.

Issue Links

• links to: GitHub Pull Request #6362
• links to: GitHub Pull Request #7489
• mentioned in: Page Loading...