feat(ingestion): Protobuf support for the Kafka source

metadata-ingestion/archived/source_docs/kafka.md

@@ -111,9 +111,9 @@ source:
 
 The Kafka Source uses the schema registry to figure out the schema associated with both `key` and `value` for the topic. 
 By default it uses the [Confluent's Kafka Schema registry](https://docs.confluent.io/platform/current/schema-registry/index.html) 
-and supports the `AVRO` schema type.
+and supports the `AVRO` and `PROTOBUF` schema types.
 
-If you're using a custom schema registry, or you are using schema type other than `AVRO`, then you can provide your own 
+If you're using a custom schema registry, or you are using schema type other than `AVRO` or `PROTOBUF`, then you can provide your own 
 custom implementation of the `KafkaSchemaRegistryBase` class, and implement the `get_schema_metadata(topic, platform_urn)` method that
 given a topic name would return object of `SchemaMetadata` containing schema for that topic. Please refer 
 `datahub.ingestion.source.confluent_schema_registry::ConfluentSchemaRegistry` for sample implementation of this class.
@@ -142,6 +142,34 @@ sink:
   # sink configs
 ```
 
+### Limitations of `PROTOBUF` schema types implementation
+
+The current implementation of the support for `PROTOBUF` schema type has the following limitations:
+
++ Recursive types are not supported.
++ If the schemas of different topics define a type in the same package, the source would raise an exception.
+
+In addition to this, maps are represented as arrays of messages. The following message,
+
+```
+message MessageWithMap {
+  map<int, string> map_1 = 1;
+}
+```
+
+becomes:
+
+```
+message Map1Entry {
+  int key = 1;
+  string value = 2/
+}
+
+message MessageWithMap {
+  repeated Map1Entry map_1 = 1;
+}
+```
+
 ## Config details
 
 Note that a `.` is used to denote nested fields in the YAML recipe.

metadata-ingestion/docs/sources/kafka/kafka.md

@@ -71,13 +71,14 @@ source:
 
 ### Custom Schema Registry
 
-The Kafka Source uses the schema registry to figure out the schema associated with both `key` and `value` for the topic. 
-By default it uses the [Confluent's Kafka Schema registry](https://docs.confluent.io/platform/current/schema-registry/index.html) 
-and supports the `AVRO` schema type.
+The Kafka Source uses the schema registry to figure out the schema associated with both `key` and `value` for the topic.
+By default it uses the [Confluent's Kafka Schema registry](https://docs.confluent.io/platform/current/schema-registry/index.html)
+and supports the `AVRO` and `PROTOBUF` schema types.
 
-If you're using a custom schema registry, or you are using schema type other than `AVRO`, then you can provide your own 
+
+If you're using a custom schema registry, or you are using schema type other than `AVRO` or `PROTOBUF`, then you can provide your own
 custom implementation of the `KafkaSchemaRegistryBase` class, and implement the `get_schema_metadata(topic, platform_urn)` method that
-given a topic name would return object of `SchemaMetadata` containing schema for that topic. Please refer 
+given a topic name would return object of `SchemaMetadata` containing schema for that topic. Please refer
 `datahub.ingestion.source.confluent_schema_registry::ConfluentSchemaRegistry` for sample implementation of this class.
 ```python
 class KafkaSchemaRegistryBase(ABC):
@@ -101,4 +102,32 @@ source:
       schema_registry_url: http://localhost:8081
 
 # sink configs
-```
+```
+
+### Limitations of `PROTOBUF` schema types implementation
+
+The current implementation of the support for `PROTOBUF` schema type has the following limitations:
+
++ Requires Python 3.7 & above.
++ Recursive types are not supported.
++ If the schemas of different topics define a type in the same package, the source would raise an exception.
+
+In addition to this, maps are represented as arrays of messages. The following message,
+
+```
+message MessageWithMap {
+  map<int, string> map_1 = 1;
+}
+```
+
+becomes:
+
+```
+message Map1Entry {
+  int key = 1;
+  string value = 2/
+}
+message MessageWithMap {
+  repeated Map1Entry map_1 = 1;
+}
+```

metadata-ingestion/setup.py

@@ -67,6 +67,20 @@ def get_long_description():
     "fastavro>=1.2.0",
 }
 
+kafka_protobuf = (
+    {
+        "networkx>=2.6.2",
+        # Required to generate protobuf python modules from the schema downloaded from the schema registry
+        "grpcio==1.44.0",
+        "grpcio-tools==1.44.0",
+        "types-protobuf",
+    }
+    if is_py37_or_newer
+    else {
+        "types-protobuf",
+    }
+)
+
 sql_common = {
     # Required for all SQL sources.
     "sqlalchemy==1.3.24",
@@ -186,7 +200,7 @@ def get_long_description():
         # - 0.6.12 adds support for Spark Thrift Server
         "acryl-pyhive[hive]>=0.6.13"
     },
-    "kafka": kafka_common,
+    "kafka": {*kafka_common, *kafka_protobuf},
     "kafka-connect": sql_common | {"requests", "JPype1"},
     "ldap": {"python-ldap>=2.4"},
     "looker": looker_common,
@@ -309,7 +323,7 @@ def get_long_description():
             "oracle",
             "postgres",
             "sagemaker",
-            "datahub-kafka",
+            "kafka",
             "datahub-rest",
             "redash",
             "redshift",