Support for camelCase

README.md

@@ -103,6 +103,14 @@ To enable this conversion mode, set the configuration field `protoMapConversionT
 value.converter.protoMapConversionType=map
 ```
 
+\* Protobuf field names are left unchanged by default, but `fieldNameConversionType` can be 
+set to `json` to use the JSON (or camel-case) field name conversion provided by the Protobuf
+library:
+ ```
+value.converter.fieldNameConversionType=json
+```
+
+
 ## Handling field renames and deletes
 Renaming and removing fields is supported by the proto IDL, but certain output formats (for example, BigQuery) do not
 support renaming or removal. In order to support these output formats, we use a custom field option to specify the

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

@@ -18,6 +18,7 @@ public class ProtobufConverter implements Converter {
   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 static final String FIELD_NAME_CONVERSION_TYPE = "fieldNameConversionType";
   private ProtobufData protobufData;
 
   private boolean isInvalidConfiguration(Object proto, boolean isKey) {
@@ -29,6 +30,7 @@ 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));
+    boolean useCamelCase = "json".equals(configs.get(FIELD_NAME_CONVERSION_TYPE));
 
     Object protoClassName = configs.get(PROTO_CLASS_NAME_CONFIG);
     if (isInvalidConfiguration(protoClassName, isKey)) {
@@ -42,8 +44,8 @@ public void configure(Map<String, ?> configs, boolean isKey) {
 
     String protoClassNameString = protoClassName.toString();
     try {
-      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);
+      log.info("Initializing ProtobufData with args: [protoClassName={}, legacyName={}, useConnectSchemaMap={}, useCamelCase={}]", protoClassNameString, legacyNameString, useConnectSchemaMap, useCamelCase);
+      protobufData = new ProtobufData(Class.forName(protoClassNameString).asSubclass(com.google.protobuf.GeneratedMessageV3.class), legacyNameString, useConnectSchemaMap, useCamelCase);
     } 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

@@ -50,6 +50,7 @@ class ProtobufData {
   private final Schema schema;
   private final String legacyName;
   private final boolean useConnectSchemaMap;
+  private final boolean useCamelCase;
   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>();
 
@@ -74,7 +75,7 @@ private String getProtoMapKey(String descriptorContainingTypeName, String connec
   }
 
   private String getConnectFieldName(Descriptors.FieldDescriptor descriptor) {
-    String name = descriptor.getName();
+    String name = useCamelCase ? descriptor.getJsonName() : descriptor.getName();
     for (Map.Entry<Descriptors.FieldDescriptor, Object> option: descriptor.getOptions().getAllFields().entrySet()) {
       if (option.getKey().getFullName().equalsIgnoreCase(this.legacyName)) {
         name = option.getValue().toString();
@@ -93,9 +94,14 @@ private String getProtoFieldName(String descriptorForTypeName, String connectFie
     this(clazz, legacyName, false);
   }
 
-  ProtobufData(Class<? extends com.google.protobuf.GeneratedMessageV3> clazz, String legacyName, boolean useConnectSchemaMap ) {
+  ProtobufData(Class<? extends com.google.protobuf.GeneratedMessageV3> clazz, String legacyName, boolean useConnectSchemaMap) {
+    this(clazz, legacyName, useConnectSchemaMap, false);
+  }
+
+  ProtobufData(Class<? extends com.google.protobuf.GeneratedMessageV3> clazz, String legacyName, boolean useConnectSchemaMap, boolean useCamelCase) {
     this.legacyName = legacyName;
     this.useConnectSchemaMap = useConnectSchemaMap;
+    this.useCamelCase = useCamelCase;
 
     try {
       this.newBuilder = clazz.getDeclaredMethod("newBuilder");

src/test/java/com/blueapron/connect/protobuf/ProtobufDataTest.java

@@ -367,6 +367,17 @@ public void testToConnectDataWithMessageWithNestedMapField() throws ParseExcepti
     assertEquals(new SchemaAndValue(getExpectedComplexMapMessageProtoSchema(true), getExpectedComplexMapMessageResult(true)), result);
   }
 
+  @Test
+  public void testToConnectDataWithMessageCamelCase() throws ParseException {
+    ComplexMapType message = createComplexMapType();
+    ProtobufData protobufData = new ProtobufData(ComplexMapType.class, LEGACY_NAME, true, true);
+    SchemaAndValue result = protobufData.toConnectData(message.toByteArray());
+    assertEquals(
+      result.schema().fields().stream().map(field -> field.name()).toArray(),
+      new String[] {"userId", "userMessages"}
+    );
+  }
+
   @Test
   public void testToConnectDataWithMessageWithNestedMapFieldListOfStruct() throws ParseException {
     ComplexMapType message = createComplexMapType();