0ct0pu5/kafka-ui
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

ISSUE-166 Pass ProtobufSchemaProvider to CachedSchemaRegistryClient to deserialize protobuf records (#178)

Browse source
This commit is contained in:
Ildar Almakaev 2021-02-10 22:28:32 +03:00 • committed by GitHub
parent 40d85643bb
commit ba4e1748ee
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 13 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/deserialization/SchemaRegistryRecordDeserializer.java
View file

@ -4,12 +4,14 @@ import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.protobuf.Message; import com.google.protobuf.Message;
import com.provectus.kafka.ui.cluster.model.KafkaCluster; 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.AvroSchemaProvider;
import io.confluent.kafka.schemaregistry.avro.AvroSchemaUtils; import io.confluent.kafka.schemaregistry.avro.AvroSchemaUtils;
import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient; import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
import io.confluent.kafka.schemaregistry.client.rest.entities.Schema; import io.confluent.kafka.schemaregistry.client.rest.entities.Schema;
import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; 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.schemaregistry.protobuf.ProtobufSchemaUtils;
import io.confluent.kafka.serializers.KafkaAvroDeserializer; import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer; import io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer;
@ -41,14 +43,17 @@ public class SchemaRegistryRecordDeserializer implements RecordDeserializer {
this.cluster = cluster; this.cluster = cluster;
this.objectMapper = objectMapper; this.objectMapper = objectMapper;
this.schemaRegistryClient = Optional.ofNullable(cluster.getSchemaRegistry()).map(e -> this.schemaRegistryClient = Optional.ofNullable(cluster.getSchemaRegistry())
new CachedSchemaRegistryClient( .map(schemaRegistryUrl -> {
Collections.singletonList(e), List<SchemaProvider> schemaProviders = List.of(new AvroSchemaProvider(), new ProtobufSchemaProvider());
CLIENT_IDENTITY_MAP_CAPACITY, return new CachedSchemaRegistryClient(
Collections.singletonList(new AvroSchemaProvider()), Collections.singletonList(schemaRegistryUrl),
Collections.emptyMap() CLIENT_IDENTITY_MAP_CAPACITY,
) schemaProviders,
).orElse(null); Collections.emptyMap()
);
}
).orElse(null);
this.avroDeserializer = Optional.ofNullable(this.schemaRegistryClient) this.avroDeserializer = Optional.ofNullable(this.schemaRegistryClient)
.map(KafkaAvroDeserializer::new) .map(KafkaAvroDeserializer::new)