Failed to sample data: Fail to get protobuf schema because of invalid schema!

[paulo-alves-flttr]

Error when using Proto bytes decoder: Failed to sample data: Fail to get protobuf schema because of invalid schema!

Affected Version

32.0.1

Description

I am trying to ingest kafka protobuf data.

I have added druid-protobuf-extensions to the druid envirionment loadlist:

druid_extensions_loadList=["druid-histogram", "druid-datasketches", "druid-lookups-cached-global", "postgresql-metadata-storage", "druid-multi-stage-query", "druid-kafka-indexing-service", "druid-avro-extensions", "druid-protobuf-extensions"]

I have added all the following jar dependencies to /opt/druid/extensions/druid-protobuf-extensions/:

kafka-protobuf-provider-6.2.12.jar
kafka-protobuf-types-6.2.12.jar
kafka-schema-registry-client-6.2.12.jar
kotlin-stdlib-common-1.4.10.jar
proto-google-common-protos-2.22.1.jar
wire-runtime-3.6.0.jar
wire-schema-3.6.0.jar

My kafka client has AUTO_REGISTER_SCHEMAS enabled and the schema is successfully created in the schema registry as following:

[
  {
    "subject": "metrics-events-value",
    "version": 1,
    "id": 1,
    "schemaType": "PROTOBUF",
    "schema": "syntax = \"proto3\";\npackage model;\n\noption java_package = \"com.flutter.kafka.proto.producer.model\";\n\nmessage Metrics {\n  string unit = 1;\n  string http_method = 2;\n  int32 value = 3;\n  string timestamp = 4;\n  string http_code = 5;\n  string page = 6;\n  string metricType = 7;\n  string server = 8;\n}\n"
  }
]

My spec is configured like this:

{
  "type": "kafka",
  "spec": {
    "ioConfig": {
      "type": "kafka",
      "consumerProperties": {
        "bootstrap.servers": "kafka:29092"
      },
      "topic": "metrics-events",
      "inputFormat": {
        "type": "kafka",
        "keyFormat": {
          "type": "regex",
          "pattern": "([\\s\\S]*)",
          "columns": [
            "line"
          ]
        },
        "valueFormat": {
          "type": "protobuf",
          "protoBytesDecoder": {
            "type": "schema_registry",
            "urls": [
              "http://schema-registry:28081"
            ],
            "capacity": 100
          }
        }
      }
    },
    "tuningConfig": {
      "type": "kafka"
    },
    "dataSchema": {
      "dataSource": "metrics-events"
    }
  }
}

Still I get "Failed to sample data: Fail to get protobuf schema because of invalid schema!" error in the UI.

I can't figure out what is missing here.

[paulo-alves-flttr]

I have discovered that the problem was that Druid 32.0.1 doesn't support the latest versions of kafka-protobuf dependencies. I have downgraded my producer to use kafka-protobuf-serializer 7.1.16 and kafka-schema-registry-maven-plugin 7.1.16 as well. Also updated my Druid image to add the following JAR files to /opt/druid/extensions/druid-protobuf-extensions/:

kafka-protobuf-provider-7.1.16.jar
kafka-protobuf-types-7.1.16.jar
kafka-schema-registry-client-7.1.16.jar
kotlin-stdlib-1.9.10.jar
okio-jvm-3.4.0.jar
proto-google-common-protos-2.22.1.jar
protobuf-java-3.25.5.jar
protobuf-java-util-3.25.5.jar
wire-runtime-jvm-4.9.7.jar
wire-schema-jvm-4.9.7.jar

This solution involved some trial and error and that could have been avoided by having supported dependencies versions listed in the documentation here.