Add support for Connect Schema Map type (blueapron#49)

* Add support for converting fields with Protobuf Map type into Connect Schema Map type

* Document the new protoMapConversionType configuration

* Log config params

Author: ofirski

README.md

@@ -87,7 +87,7 @@ key.converter=com.blueapron.connect.protobuf.ProtobufConverter
 |string       |STRING             |
 |bytes        |BYTES              |
 |repeated     |ARRAY              |
-|map          |ARRAY of STRUCT    |
+|map          |ARRAY of STRUCT*   |
 
 
 |Protobuf Type|Connect Logical Type|
@@ -96,6 +96,12 @@ key.converter=com.blueapron.connect.protobuf.ProtobufConverter
 |Date         |Date                |
 |uint64       |Decimal             |
 
+\* Protobuf map type fields are converted by default to an Array of Struct, 
+but the converter supports also converting these fields into Connect Schema Map type.
+To enable this conversion mode, set the configuration field `protoMapConversionType` to `map`:
+ ```
+value.converter.protoMapConversionType=map
+```
 
 ## Handling field renames and deletes
 Renaming and removing fields is supported by the proto IDL, but certain output formats (for example, BigQuery) do not

src/main/java/com/blueapron/connect/protobuf/ProtobufConverter.java

@@ -17,6 +17,7 @@ public class ProtobufConverter implements Converter {
   private static final Logger log = LoggerFactory.getLogger(ProtobufConverter.class);
   private static final String PROTO_CLASS_NAME_CONFIG = "protoClassName";
   private static final String LEGACY_NAME_CONFIG = "legacyName";
+  private static final String PROTO_MAP_CONVERSION_TYPE = "protoMapConversionType";
   private ProtobufData protobufData;
 
   private boolean isInvalidConfiguration(Object proto, boolean isKey) {
@@ -27,6 +28,7 @@ private boolean isInvalidConfiguration(Object proto, boolean isKey) {
   public void configure(Map<String, ?> configs, boolean isKey) {
     Object legacyName = configs.get(LEGACY_NAME_CONFIG);
     String legacyNameString = legacyName == null ? "legacy_name" : legacyName.toString();
+    boolean useConnectSchemaMap = "map".equals(configs.get(PROTO_MAP_CONVERSION_TYPE));
 
     Object protoClassName = configs.get(PROTO_CLASS_NAME_CONFIG);
     if (isInvalidConfiguration(protoClassName, isKey)) {
@@ -40,7 +42,8 @@ public void configure(Map<String, ?> configs, boolean isKey) {
 
     String protoClassNameString = protoClassName.toString();
     try {
-      protobufData = new ProtobufData(Class.forName(protoClassNameString).asSubclass(com.google.protobuf.GeneratedMessageV3.class), legacyNameString);
+      log.info("Initializing ProtobufData with args: [protoClassName={}, legacyName={}, useConnectSchemaMap={}]", protoClassNameString, legacyNameString, useConnectSchemaMap);
+      protobufData = new ProtobufData(Class.forName(protoClassNameString).asSubclass(com.google.protobuf.GeneratedMessageV3.class), legacyNameString, useConnectSchemaMap);
     } catch (ClassNotFoundException e) {
       throw new ConnectException("Proto class " + protoClassNameString + " not found in the classpath");
     } catch (ClassCastException e) {

src/main/java/com/blueapron/connect/protobuf/ProtobufData.java

@@ -7,6 +7,7 @@
 import com.google.protobuf.Descriptors.OneofDescriptor;
 import com.google.protobuf.GeneratedMessageV3;
 import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.MapEntry;
 import com.google.protobuf.Message;
 import org.apache.kafka.connect.data.Date;
 import org.apache.kafka.connect.data.Decimal;
@@ -28,6 +29,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 import com.google.protobuf.util.Timestamps;
 
 import static com.google.protobuf.Descriptors.FieldDescriptor.Type.BOOL;
@@ -47,6 +49,7 @@ class ProtobufData {
   private final Method newBuilder;
   private final Schema schema;
   private final String legacyName;
+  private final boolean useConnectSchemaMap;
   public static final Descriptors.FieldDescriptor.Type[] PROTO_TYPES_WITH_DEFAULTS = new Descriptors.FieldDescriptor.Type[] { INT32, INT64, SINT32, SINT64, FLOAT, DOUBLE, BOOL, STRING, BYTES, ENUM };
   private HashMap<String, String> connectProtoNameMap = new HashMap<String, String>();
 
@@ -87,7 +90,12 @@ private String getProtoFieldName(String descriptorForTypeName, String connectFie
   }
 
   ProtobufData(Class<? extends com.google.protobuf.GeneratedMessageV3> clazz, String legacyName) {
+    this(clazz, legacyName, false);
+  }
+
+  ProtobufData(Class<? extends com.google.protobuf.GeneratedMessageV3> clazz, String legacyName, boolean useConnectSchemaMap ) {
     this.legacyName = legacyName;
+    this.useConnectSchemaMap = useConnectSchemaMap;
 
     try {
       this.newBuilder = clazz.getDeclaredMethod("newBuilder");
@@ -194,6 +202,14 @@ private Schema toConnectSchema(Descriptors.FieldDescriptor descriptor) {
           builder = Date.builder();
           break;
         }
+
+        if (shouldConvertToConnectSchemaMap(descriptor)) {
+          FieldDescriptor keyFieldDescriptor = descriptor.getMessageType().findFieldByName("key");
+          FieldDescriptor valueFieldDescriptor = descriptor.getMessageType().findFieldByName("value");
+          builder = SchemaBuilder.map(toConnectSchema(keyFieldDescriptor), toConnectSchema(valueFieldDescriptor));
+          break;
+        }
+
         String jsonName = descriptor.getJsonName();
         builder = SchemaBuilder.struct().name(jsonName.substring(0, 1).toUpperCase() + jsonName.substring(1));
         for (Descriptors.FieldDescriptor fieldDescriptor : descriptor.getMessageType().getFields()) {
@@ -210,7 +226,7 @@ private Schema toConnectSchema(Descriptors.FieldDescriptor descriptor) {
     builder.optional();
     Schema schema = builder.build();
 
-    if (descriptor.isRepeated()) {
+    if (descriptor.isRepeated() && !shouldConvertToConnectSchemaMap(descriptor)) {
       final SchemaBuilder arrayBuilder = SchemaBuilder.array(schema);
       arrayBuilder.optional();
       schema = arrayBuilder.build();
@@ -219,6 +235,10 @@ private Schema toConnectSchema(Descriptors.FieldDescriptor descriptor) {
     return schema;
   }
 
+  private boolean shouldConvertToConnectSchemaMap(FieldDescriptor descriptor) {
+    return useConnectSchemaMap && descriptor.isMapField();
+  }
+
   private boolean isProtobufTimestamp(Schema schema) {
     return Timestamp.SCHEMA.name().equals(schema.name());
   }
@@ -340,6 +360,15 @@ Object toConnectData(Schema schema, Object value) {
           break;
         }
 
+        case MAP: {
+          Collection<MapEntry> original = (Collection<MapEntry>) value;
+          converted = original.stream().collect(Collectors.toMap(
+            entry -> toConnectData(schema.keySchema(), entry.getKey()),
+            entry -> toConnectData(schema.valueSchema(), entry.getValue())
+          ));
+          break;
+        }
+
         case STRUCT: {
           final Message message = (Message) value; // Validate type