(WIP) KAFKA-5142: Expose Record Headers in Kafka Connect (DO NOT MERGE)

clients/src/main/java/org/apache/kafka/common/serialization/BooleanDeserializer.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.serialization;
+
+import java.util.Map;
+
+import org.apache.kafka.common.errors.SerializationException;
+
+public class BooleanDeserializer implements Deserializer<Boolean> {
+
+    public void configure(Map<String, ?> configs, boolean isKey) {
+        // nothing to do
+    }
+
+    public Boolean deserialize(String topic, byte[] data) {
+        if (data == null)
+            return null;
+        if (data.length != 1) {
+            throw new SerializationException("Size of data received by IntegerDeserializer is " +
+                    "not 1");
+        }
+        return data[0] == 1;
+    }
+
+    public void close() {
+        // nothing to do
+    }
+}

clients/src/main/java/org/apache/kafka/common/serialization/BooleanSerializer.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.serialization;
+
+import java.util.Map;
+
+public class BooleanSerializer implements Serializer<Boolean> {
+
+    public void configure(Map<String, ?> configs, boolean isKey) {
+        // nothing to do
+    }
+
+    public byte[] serialize(String topic, Boolean data) {
+        if (data == null)
+            return null;
+
+        byte b = (byte)(data?1:0);
+        return new byte[] {b};
+    }
+
+    public void close() {
+        // nothing to do
+    }
+}

clients/src/main/java/org/apache/kafka/common/serialization/ByteDeserializer.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.serialization;
+
+import java.util.Map;
+
+import org.apache.kafka.common.errors.SerializationException;
+
+public class ByteDeserializer implements Deserializer<Byte> {
+
+    public void configure(Map<String, ?> configs, boolean isKey) {
+        // nothing to do
+    }
+
+    public Byte deserialize(String topic, byte[] data) {
+        if (data == null)
+            return null;
+        if (data.length != 1) {
+            throw new SerializationException("Size of data received by IntegerDeserializer is " +
+                    "not 1");
+        }
+        return data[0];
+    }
+
+    public void close() {
+        // nothing to do
+    }
+}

clients/src/main/java/org/apache/kafka/common/serialization/ByteSerializer.java

@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.serialization;
+
+import java.util.Map;
+
+public class ByteSerializer implements Serializer<Byte> {
+
+    public void configure(Map<String, ?> configs, boolean isKey) {
+        // nothing to do
+    }
+
+    public byte[] serialize(String topic, Byte data) {
+        if (data == null)
+            return null;
+
+        return new byte[] {data};
+    }
+
+    public void close() {
+        // nothing to do
+    }
+}

clients/src/main/java/org/apache/kafka/common/serialization/ShortDeserializer.java

@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.serialization;
+
+import java.util.Map;
+
+import org.apache.kafka.common.errors.SerializationException;
+
+public class ShortDeserializer implements Deserializer<Short> {
+
+    public void configure(Map<String, ?> configs, boolean isKey) {
+        // nothing to do
+    }
+
+    public Short deserialize(String topic, byte[] data) {
+        if (data == null)
+            return null;
+        if (data.length != 2) {
+            throw new SerializationException("Size of data received by IntegerDeserializer is " +
+                    "not 2");
+        }
+
+        short value = 0;
+        for (byte b : data) {
+            value <<= 8;
+            value |= b & 0xFF;
+        }
+        return value;
+    }
+
+    public void close() {
+        // nothing to do
+    }
+}

clients/src/main/java/org/apache/kafka/common/serialization/ShortSerializer.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.serialization;
+
+import java.util.Map;
+
+public class ShortSerializer implements Serializer<Short> {
+
+    public void configure(Map<String, ?> configs, boolean isKey) {
+        // nothing to do
+    }
+
+    public byte[] serialize(String topic, Short data) {
+        if (data == null)
+            return null;
+
+        return new byte[] {
+            (byte) (data >>> 8),
+            data.byteValue()
+        };
+    }
+
+    public void close() {
+        // nothing to do
+    }
+}

connect/api/src/main/java/org/apache/kafka/connect/connector/ConnectHeader.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.connector;
+
+import org.apache.kafka.connect.data.Schema;
+
+/**
+ * <p>
+ * Class for headers containing data to be copied to/from Kafka. This corresponds closely to
+ * Kafka's Header classes, and holds the data that may be used by both
+ * sources and sinks (key, valueSchema, value).
+ * </p>
+ */
+public class ConnectHeader {
+
+    private final String key;
+    private final Schema valueSchema;
+    private final Object value;
+
+    public ConnectHeader(String key, Schema valueSchema, Object value) {
+        this.key = key;
+        this.valueSchema = valueSchema;
+        this.value = value;
+    }
+
+    public String key() {
+        return key;
+    }
+
+    public Object value() {
+        return value;
+    }
+
+    public Schema valueSchema() {
+        return valueSchema;
+    }
+
+}

connect/api/src/main/java/org/apache/kafka/connect/connector/ConnectRecord.java

@@ -16,6 +16,9 @@
  */
 package org.apache.kafka.connect.connector;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import org.apache.kafka.connect.data.Schema;
 
 /**
@@ -34,18 +37,20 @@ public abstract class ConnectRecord<R extends ConnectRecord<R>> {
     private final Schema valueSchema;
     private final Object value;
     private final Long timestamp;
+    private final List<ConnectHeader> headers;
 
     public ConnectRecord(String topic, Integer kafkaPartition,
                          Schema keySchema, Object key,
                          Schema valueSchema, Object value,
-                         Long timestamp) {
+                         Long timestamp, List<ConnectHeader> headers) {
         this.topic = topic;
         this.kafkaPartition = kafkaPartition;
         this.keySchema = keySchema;
         this.key = key;
         this.valueSchema = valueSchema;
         this.value = value;
         this.timestamp = timestamp;
+        this.headers = headers == null ? new ArrayList<ConnectHeader>() : new ArrayList<>(headers);
     }
 
     public String topic() {
@@ -71,13 +76,22 @@ public Object value() {
     public Schema valueSchema() {
         return valueSchema;
     }
+
+    public List<ConnectHeader> headers() {
+        return headers;
+    }
 
     public Long timestamp() {
         return timestamp;
     }
 
     /** Generate a new record of the same type as itself, with the specified parameter values. **/
-    public abstract R newRecord(String topic, Integer kafkaPartition, Schema keySchema, Object key, Schema valueSchema, Object value, Long timestamp);
+    public R newRecord(String topic, Integer kafkaPartition, Schema keySchema, Object key, Schema valueSchema, Object value, Long timestamp) {
+        return newRecord(topic, kafkaPartition, keySchema, key, valueSchema, value, timestamp, this.headers());
+    }
+
+    /** Generate a new record of the same type as itself, with the specified parameter values. **/
+    public abstract R newRecord(String topic, Integer kafkaPartition, Schema keySchema, Object key, Schema valueSchema, Object value, Long timestamp, List<ConnectHeader> headers);
 
     @Override
     public String toString() {
@@ -87,7 +101,8 @@ public String toString() {
                 ", key=" + key +
                 ", value=" + value +
                 ", timestamp=" + timestamp +
-                '}';
+                ", headers=" + headers +
+               '}';
     }
 
     @Override
@@ -113,6 +128,8 @@ public boolean equals(Object o) {
             return false;
         if (timestamp != null ? !timestamp.equals(that.timestamp) : that.timestamp != null)
             return false;
+        if (!headers.equals(that.headers))
+            return false;
 
         return true;
     }
@@ -126,6 +143,7 @@ public int hashCode() {
         result = 31 * result + (valueSchema != null ? valueSchema.hashCode() : 0);
         result = 31 * result + (value != null ? value.hashCode() : 0);
         result = 31 * result + (timestamp != null ? timestamp.hashCode() : 0);
+        result = 31 * result + headers.hashCode();
         return result;
     }
 }