Repository: provectus/kafka-ui
Issue: Protobuf not working
Labels: type/bug
scope/backend

Hi,

Really interesting project :wink:  

#116 seems to add protobuf supports but when I try to read message serialized with protobuf I get this error message:
```
ERROR io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient - Invalid schema type PROTOBUF
```

Full stacktrace:
```
11:38:45.932 [boundedElastic-2] INFO  org.apache.kafka.clients.consumer.ConsumerConfig - ConsumerConfig values: 
	allow.auto.create.topics = true
	auto.commit.interval.ms = 5000
	auto.offset.reset = earliest
	bootstrap.servers = [kafka:9092]
	check.crcs = true
	client.dns.lookup = default
	client.id = kafka-ui
	client.rack = 
	connections.max.idle.ms = 540000
	default.api.timeout.ms = 60000
	enable.auto.commit = true
	exclude.internal.topics = true
	fetch.max.bytes = 52428800
	fetch.max.wait.ms = 500
	fetch.min.bytes = 1
	group.id = null
	group.instance.id = null
	heartbeat.interval.ms = 3000
	interceptor.classes = []
	internal.leave.group.on.close = true
	isolation.level = read_uncommitted
	key.deserializer = class org.apache.kafka.common.serialization.BytesDeserializer
	max.partition.fetch.bytes = 1048576
	max.poll.interval.ms = 300000
	max.poll.records = 500
	metadata.max.age.ms = 300000
	metric.reporters = []
	metrics.num.samples = 2
	metrics.recording.level = INFO
	metrics.sample.window.ms = 30000
	partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
	receive.buffer.bytes = 65536
	reconnect.backoff.max.ms = 1000
	reconnect.backoff.ms = 50
	request.timeout.ms = 30000
	retry.backoff.ms = 100
	sasl.client.callback.handler.class = null
	sasl.jaas.config = null
	sasl.kerberos.kinit.cmd = /usr/bin/kinit
	sasl.kerberos.min.time.before.relogin = 60000
	sasl.kerberos.service.name = null
	sasl.kerberos.ticket.renew.jitter = 0.05
	sasl.kerberos.ticket.renew.window.factor = 0.8
	sasl.login.callback.handler.class = null
	sasl.login.class = null
	sasl.login.refresh.buffer.seconds = 300
	sasl.login.refresh.min.period.seconds = 60
	sasl.login.refresh.window.factor = 0.8
	sasl.login.refresh.window.jitter = 0.05
	sasl.mechanism = GSSAPI
	security.protocol = PLAINTEXT
	security.providers = null
	send.buffer.bytes = 131072
	session.timeout.ms = 10000
	ssl.cipher.suites = null
	ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
	ssl.endpoint.identification.algorithm = https
	ssl.key.password = null
	ssl.keymanager.algorithm = SunX509
	ssl.keystore.location = null
	ssl.keystore.password = null
	ssl.keystore.type = JKS
	ssl.protocol = TLS
	ssl.provider = null
	ssl.secure.random.implementation = null
	ssl.trustmanager.algorithm = PKIX
	ssl.truststore.location = null
	ssl.truststore.password = null
	ssl.truststore.type = JKS
	value.deserializer = class org.apache.kafka.common.serialization.BytesDeserializer

11:38:45.937 [boundedElastic-2] INFO  org.apache.kafka.common.utils.AppInfoParser - Kafka version: 2.4.0
11:38:45.937 [boundedElastic-2] INFO  org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: 77a89fcf8d7fa018
11:38:45.937 [boundedElastic-2] INFO  org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: 1611920325937
11:38:45.938 [boundedElastic-2] INFO  org.apache.kafka.clients.consumer.KafkaConsumer - [Consumer clientId=kafka-ui, groupId=null] Subscribed to partition(s): local-states-0
11:38:45.940 [boundedElastic-2] INFO  org.apache.kafka.clients.consumer.internals.SubscriptionState - [Consumer clientId=kafka-ui, groupId=null] Seeking to EARLIEST offset of partition local-states-0
11:38:45.940 [boundedElastic-2] INFO  com.provectus.kafka.ui.cluster.service.ConsumingService - assignment: [local-states-0]
11:38:45.946 [boundedElastic-2] INFO  org.apache.kafka.clients.Metadata - [Consumer clientId=kafka-ui, groupId=null] Cluster ID: nrHf2NLFSV6lTctErdFmCQ
11:38:45.950 [boundedElastic-2] INFO  org.apache.kafka.clients.consumer.internals.SubscriptionState - [Consumer clientId=kafka-ui, groupId=null] Resetting offset for partition local-states-0 to offset 3074.
11:38:45.975 [boundedElastic-2] INFO  com.provectus.kafka.ui.cluster.service.ConsumingService - 500 records polled
11:38:46.037 [boundedElastic-2] ERROR io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient - Invalid schema type PROTOBUF
11:38:46.037 [boundedElastic-2] INFO  com.provectus.kafka.ui.cluster.deserialization.SchemaRegistryRecordDeserializer - Failed to get Protobuf schema for topic local-states
org.apache.kafka.common.errors.SerializationException: Error deserializing Protobuf message for id 6
Caused by: java.io.IOException: Invalid schema syntax = "proto3";
package io.spoud.sdm.hooks.domain.v1;

import "google/protobuf/timestamp.proto";
import "logistics/domain/v1/domain.proto";
import "global/domain/v1/domain.proto";

option java_package = "io.spoud.sdm.hooks.domain.v1";
option java_multiple_files = true;

message LogRecord {
  string entity_uuid = 1;
  .google.protobuf.Timestamp timestamp = 2;
  .io.spoud.sdm.hooks.domain.v1.StateChangeAction.Type action = 3;
  .io.spoud.sdm.global.domain.v1.ResourceEntity.Type entity_type = 4;
  string path = 5;
  string cursor = 6;

  oneof entity {
    .io.spoud.sdm.logistics.domain.v1.DataOffer data_offer = 10;
    .io.spoud.sdm.logistics.domain.v1.DataPort data_port = 12;
    .io.spoud.sdm.logistics.domain.v1.DataSubscriptionState data_subscription_state = 14;
    .io.spoud.sdm.logistics.domain.v1.ResourceGroup resource_group = 15;
    .io.spoud.sdm.logistics.domain.v1.Transport transport = 16;
    .io.spoud.sdm.logistics.domain.v1.DataItem data_item = 17;
  }
}
message StreamInfo {
  .google.protobuf.Timestamp timestamp = 2;
}
message StateChangeAction {
  enum Type {
    UNKNOWN = 0;
    UPDATED = 1;
    DELETED = 2;
  }
}
 with refs [{name='google/protobuf/timestamp.proto', subject='google/protobuf/timestamp.proto', version=1}, {name='logistics/domain/v1/domain.proto', subject='logistics/domain/v1/domain.proto', version=1}, {name='global/domain/v1/domain.proto', subject='global/domain/v1/domain.proto', version=1}] of type PROTOBUF
	at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.lambda$getSchemaByIdFromRegistry$4(CachedSchemaRegistryClient.java:222) ~[kafka-schema-registry-client-5.5.1.jar!/:?]
	at java.util.Optional.orElseThrow(Optional.java:401) ~[?:?]
	at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaByIdFromRegistry(CachedSchemaRegistryClient.java:220) ~[kafka-schema-registry-client-5.5.1.jar!/:?]
	at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaBySubjectAndId(CachedSchemaRegistryClient.java:291) ~[kafka-schema-registry-client-5.5.1.jar!/:?]
	at io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient.getSchemaById(CachedSchemaRegistryClient.java:276) ~[kafka-schema-registry-client-5.5.1.jar!/:?]
	at io.confluent.kafka.serializers.protobuf.AbstractKafkaProtobufDeserializer.deserialize(AbstractKafkaProtobufDeserializer.java:105) ~[kafka-protobuf-serializer-5.5.1.jar!/:?]
	at io.confluent.kafka.serializers.protobuf.AbstractKafkaProtobufDeserializer.deserialize(AbstractKafkaProtobufDeserializer.java:86) ~[kafka-protobuf-serializer-5.5.1.jar!/:?]
	at io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer.deserialize(KafkaProtobufDeserializer.java:75) ~[kafka-protobuf-serializer-5.5.1.jar!/:?]
	at com.provectus.kafka.ui.cluster.deserialization.SchemaRegistryRecordDeserializer.detectFormat(SchemaRegistryRecordDeserializer.java:104) ~[classes!/:?]
	at com.provectus.kafka.ui.cluster.deserialization.SchemaRegistryRecordDeserializer.lambda$getMessageFormat$1(SchemaRegistryRecordDeserializer.java:90) ~[classes!/:?]
	at java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1705) ~[?:?]
	at com.provectus.kafka.ui.cluster.deserialization.SchemaRegistryRecordDeserializer.getMessageFormat(SchemaRegistryRecordDeserializer.java:90) ~[classes!/:?]
	at com.provectus.kafka.ui.cluster.deserialization.SchemaRegistryRecordDeserializer.deserialize(SchemaRegistryRecordDeserializer.java:63) ~[classes!/:?]
	at com.provectus.kafka.ui.cluster.util.ClusterUtil.mapToTopicMessage(ClusterUtil.java:171) ~[classes!/:?]
	at com.provectus.kafka.ui.cluster.service.ConsumingService.lambda$loadMessages$1(ConsumingService.java:53) ~[classes!/:?]
	at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:199) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
	at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.onNext(FluxSubscribeOn.java:151) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
	at reactor.core.publisher.FluxCreate$BufferAsyncSink.drain(FluxCreate.java:793) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
	at reactor.core.publisher.FluxCreate$BufferAsyncSink.next(FluxCreate.java:718) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
	at reactor.core.publisher.FluxCreate$SerializedSink.next(FluxCreate.java:153) ~[reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
	at java.util.Iterator.forEachRemaining(Iterator.java:133) ~[?:?]
	at com.provectus.kafka.ui.cluster.service.ConsumingService$RecordEmitter.emit(ConsumingService.java:113) ~[classes!/:?]
	at reactor.core.publisher.FluxCreate.subscribe(FluxCreate.java:94) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
	at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) [reactor-core-3.3.2.RELEASE.jar!/:3.3.2.RELEASE]
	at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) [?:?]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
	at java.lang.Thread.run(Thread.java:830) [?:?]
11:38:46.044 [boundedElastic-2] INFO  com.provectus.kafka.ui.cluster.deserialization.SchemaRegistryRecordDeserializer - Failed to parse json from topic local-states
11:38:46.056 [boundedElastic-2] DEBUG org.springframework.http.codec.json.Jackson2JsonEncoder - [c3c611fb] Encoding [[class TopicMessage {
    partition: 0
    offset: 3074
    timestamp: 2021-01-25T07:42:24.057Z
     (truncated)...]

```

Issue comments:
Thank your for your issue, We'll try to fix it soon.

Pull request review comments:

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/deserialization/SchemaRegistryRecordDeserializer.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/deserialization/SchemaRegistryRecordDeserializer.java
index a64aded30a7..00544a109dc 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/deserialization/SchemaRegistryRecordDeserializer.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/deserialization/SchemaRegistryRecordDeserializer.java
@@ -4,12 +4,14 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.protobuf.Message;
 import com.provectus.kafka.ui.cluster.model.KafkaCluster;
+import io.confluent.kafka.schemaregistry.SchemaProvider;
 import io.confluent.kafka.schemaregistry.avro.AvroSchemaProvider;
 import io.confluent.kafka.schemaregistry.avro.AvroSchemaUtils;
 import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
 import io.confluent.kafka.schemaregistry.client.rest.entities.Schema;
 import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
+import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchemaProvider;
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchemaUtils;
 import io.confluent.kafka.serializers.KafkaAvroDeserializer;
 import io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer;
@@ -41,14 +43,17 @@ public SchemaRegistryRecordDeserializer(KafkaCluster cluster, ObjectMapper objec
 		this.cluster = cluster;
 		this.objectMapper = objectMapper;
 
-		this.schemaRegistryClient = Optional.ofNullable(cluster.getSchemaRegistry()).map(e ->
-						new CachedSchemaRegistryClient(
-								Collections.singletonList(e),
-								CLIENT_IDENTITY_MAP_CAPACITY,
-								Collections.singletonList(new AvroSchemaProvider()),
-								Collections.emptyMap()
-						)
-		).orElse(null);
+		this.schemaRegistryClient = Optional.ofNullable(cluster.getSchemaRegistry())
+                .map(schemaRegistryUrl -> {
+                            List<SchemaProvider> schemaProviders = List.of(new AvroSchemaProvider(), new ProtobufSchemaProvider());
+                            return new CachedSchemaRegistryClient(
+                                    Collections.singletonList(schemaRegistryUrl),
+                                    CLIENT_IDENTITY_MAP_CAPACITY,
+                                    schemaProviders,
+                                    Collections.emptyMap()
+                            );
+                        }
+                ).orElse(null);
 
 		this.avroDeserializer = Optional.ofNullable(this.schemaRegistryClient)
 				.map(KafkaAvroDeserializer::new)


Test patch: