add protobuf inputformat

docs/ingestion/data-formats.md

@@ -272,6 +272,41 @@ The `inputFormat` to load data of Avro OCF format. An example is:
 |schema| JSON Object |Define a reader schema to be used when parsing Avro records, this is useful when parsing multiple versions of Avro OCF file data | no (default will decode using the writer schema contained in the OCF file) |
 | binaryAsString | Boolean | Specifies if the bytes parquet column which is not logically marked as a string or enum type should be treated as a UTF-8 encoded string. | no (default = false) |
 
+### Protobuf
+
+> You need to include the [`druid-protobuf-extensions`](../development/extensions-core/protobuf.md) as an extension to use the Protobuf input format.
+
+The `inputFormat` to load data of Protobuf format. An example is:
+```json
+"ioConfig": {
+  "inputFormat": {
+    "type": "protobuf",
+    "protoBytesDecoder": {
+      "type": "file",
+      "descriptor": "file:///tmp/metrics.desc",
+      "protoMessageType": "Metrics"
+    }
+    "flattenSpec": {
+      "useFieldDiscovery": true,
+      "fields": [
+        {
+          "type": "path",
+          "name": "someRecord_subInt",
+          "expr": "$.someRecord.subInt"
+        }
+      ]
+    }
+  },
+  ...
+}
+```
+
+| Field | Type | Description | Required |
+|-------|------|-------------|----------|
+|type| String| This should be set to `protobuf` to read Protobuf serialized data| yes |
+|flattenSpec| JSON Object |Define a [`flattenSpec`](#flattenspec) to extract nested values from a Protobuf record. Note that only 'path' expression are supported ('jq' is unavailable).| no (default will auto-discover 'root' level properties) |
+|`protoBytesDecoder`| JSON Object |Specifies how to decode bytes to Protobuf record. | yes |
+
 ### FlattenSpec
 
 The `flattenSpec` is located in `inputFormat` → `flattenSpec` and is responsible for
@@ -1066,17 +1101,19 @@ This parser is for [stream ingestion](./index.md#streaming) and reads Protocol b
 | Field | Type | Description | Required |
 |-------|------|-------------|----------|
 | type | String | This should say `protobuf`. | yes |
-| descriptor | String | Protobuf descriptor file name in the classpath or URL. | yes |
-| protoMessageType | String | Protobuf message type in the descriptor.  Both short name and fully qualified name are accepted.  The parser uses the first message type found in the descriptor if not specified. | no |
+| `protoBytesDecoder` | JSON Object | Specifies how to decode bytes to Protobuf record. | yes |
 | parseSpec | JSON Object | Specifies the timestamp and dimensions of the data.  The format must be JSON. See [JSON ParseSpec](./index.md) for more configuration options.  Note that timeAndDims parseSpec is no longer supported. | yes |
 
 Sample spec:
 
 ```json
 "parser": {
   "type": "protobuf",
-  "descriptor": "file:///tmp/metrics.desc",
-  "protoMessageType": "Metrics",
+  "protoBytesDecoder": {
+    "type": "file",
+    "descriptor": "file:///tmp/metrics.desc",
+    "protoMessageType": "Metrics"
+  },
   "parseSpec": {
     "format": "json",
     "timestampSpec": {
@@ -1104,6 +1141,83 @@ Sample spec:
 See the [extension description](../development/extensions-core/protobuf.md) for
 more details and examples.
 
+#### Protobuf Bytes Decoder
+
+If `type` is not included, the `protoBytesDecoder` defaults to `schema_registry`.
+
+##### File-based Protobuf Bytes Decoder
+
+This Protobuf bytes decoder first read a descriptor file, and then parse it to get schema used to decode the Protobuf record from bytes.
+
+| Field | Type | Description | Required |
+|-------|------|-------------|----------|
+| type | String | This should say `file`. | yes |
+| descriptor | String | Protobuf descriptor file name in the classpath or URL. | yes |
+| protoMessageType | String | Protobuf message type in the descriptor.  Both short name and fully qualified name are accepted.  The parser uses the first message type found in the descriptor if not specified. | no |
+
+Sample spec:
+
+```json
+"protoBytesDecoder": {
+  "type": "file",
+  "descriptor": "file:///tmp/metrics.desc",
+  "protoMessageType": "Metrics"
+}
+```
+
+##### Confluent Schema Registry-based Protobuf Bytes Decoder
+
+This Protobuf bytes decoder first extracts a unique `id` from input message bytes, and then uses it to look up the schema in the Schema Registry used to decode the Avro record from bytes.
+For details, see the Schema Registry [documentation](http://docs.confluent.io/current/schema-registry/docs/) and [repository](https://github.com/confluentinc/schema-registry).
+
+| Field | Type | Description | Required |
+|-------|------|-------------|----------|
+| type | String | This should say `schema_registry`. | yes |
+| url | String | Specifies the url endpoint of the Schema Registry. | yes |
+| capacity | Integer | Specifies the max size of the cache (default = Integer.MAX_VALUE). | no |
+| urls | Array<String> | Specifies the url endpoints of the multiple Schema Registry instances. | yes(if `url` is not provided) |
+| config | Json | To send additional configurations, configured for Schema Registry | no |
+| headers | Json | To send headers to the Schema Registry | no |
+
+For a single schema registry instance, use Field `url` or `urls` for multi instances.
+
+Single Instance:
+
+```json
+...
+"protoBytesDecoder": {
+  "url": <schema-registry-url>,
+  "type": "schema_registry"
+}
+...
+```
+
+Multiple Instances:
+```json
+...
+"protoBytesDecoder": {
+  "urls": [<schema-registry-url-1>, <schema-registry-url-2>, ...],
+  "type": "schema_registry",
+  "capacity": 100,
+  "config" : {
+       "basic.auth.credentials.source": "USER_INFO",
+       "basic.auth.user.info": "fred:letmein",
+       "schema.registry.ssl.truststore.location": "/some/secrets/kafka.client.truststore.jks",
+       "schema.registry.ssl.truststore.password": "<password>",
+       "schema.registry.ssl.keystore.location": "/some/secrets/kafka.client.keystore.jks",
+       "schema.registry.ssl.keystore.password": "<password>",
+       "schema.registry.ssl.key.password": "<password>",
+         ... 
+  },
+  "headers": {
+      "traceID" : "b29c5de2-0db4-490b-b421",
+      "timeStamp" : "1577191871865",
+      ...
+  }
+}
+...
+```
+
 ## ParseSpec
 
 > The Parser is deprecated for [native batch tasks](./native-batch.md), [Kafka indexing service](../development/extensions-core/kafka-ingestion.md),

extensions-core/protobuf-extensions/pom.xml

@@ -137,6 +137,10 @@
       <version>2.0.1</version>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-core</artifactId>
+    </dependency>
 
     <!-- test -->
     <dependency>
@@ -154,6 +158,12 @@
       <artifactId>mockito-core</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.druid</groupId>
+      <artifactId>druid-processing</artifactId>
+      <version>${project.parent.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoder.java

@@ -34,6 +34,7 @@
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.nio.ByteBuffer;
+import java.util.Objects;
 import java.util.Set;
 
 public class FileBasedProtobufBytesDecoder implements ProtobufBytesDecoder
@@ -54,6 +55,18 @@ public FileBasedProtobufBytesDecoder(
     initDescriptor();
   }
 
+  @JsonProperty
+  public String getDescriptor()
+  {
+    return descriptorFilePath;
+  }
+
+  @JsonProperty
+  public String getProtoMessageType()
+  {
+    return protoMessageType;
+  }
+
   @VisibleForTesting
   void initDescriptor()
   {
@@ -123,4 +136,27 @@ private Descriptors.Descriptor getDescriptor(String descriptorFilePath)
     }
     return desc;
   }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    FileBasedProtobufBytesDecoder that = (FileBasedProtobufBytesDecoder) o;
+
+    return Objects.equals(descriptorFilePath, that.descriptorFilePath) &&
+        Objects.equals(protoMessageType, that.protoMessageType);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(descriptorFilePath, protoMessageType);
+  }
+
 }

extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/ProtobufExtensionsModule.java

@@ -37,7 +37,8 @@ public List<? extends Module> getJacksonModules()
     return Collections.singletonList(
         new SimpleModule("ProtobufInputRowParserModule")
             .registerSubtypes(
-                new NamedType(ProtobufInputRowParser.class, "protobuf")
+                new NamedType(ProtobufInputRowParser.class, "protobuf"),
+                new NamedType(ProtobufInputFormat.class, "protobuf")
             )
     );
   }

extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/ProtobufInputFormat.java

@@ -0,0 +1,92 @@
+/*
+ * 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.druid.data.input.protobuf;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.data.input.InputEntity;
+import org.apache.druid.data.input.InputEntityReader;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.impl.NestedInputFormat;
+import org.apache.druid.java.util.common.parsers.JSONPathSpec;
+
+import javax.annotation.Nullable;
+
+import java.io.File;
+import java.util.Objects;
+
+public class ProtobufInputFormat extends NestedInputFormat
+{
+  private final ProtobufBytesDecoder protobufBytesDecoder;
+
+  @JsonCreator
+  public ProtobufInputFormat(
+      @JsonProperty("flattenSpec") @Nullable JSONPathSpec flattenSpec,
+      @JsonProperty("protoBytesDecoder") ProtobufBytesDecoder protobufBytesDecoder
+  )
+  {
+    super(flattenSpec);
+    this.protobufBytesDecoder = protobufBytesDecoder;
+  }
+
+  @JsonProperty
+  public ProtobufBytesDecoder getProtoBytesDecoder()
+  {
+    return protobufBytesDecoder;
+  }
+
+  @Override
+  public boolean isSplittable()
+  {
+    return false;
+  }
+
+  @Override
+  public InputEntityReader createReader(InputRowSchema inputRowSchema, InputEntity source, File temporaryDirectory)
+  {
+    return new ProtobufReader(
+        inputRowSchema,
+        source,
+        protobufBytesDecoder,
+        getFlattenSpec()
+    );
+  }
+
+  @Override
+  public boolean equals(final Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    final ProtobufInputFormat that = (ProtobufInputFormat) o;
+    return Objects.equals(getFlattenSpec(), that.getFlattenSpec()) &&
+        Objects.equals(protobufBytesDecoder, that.protobufBytesDecoder);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(protobufBytesDecoder, getFlattenSpec());
+  }
+
+}