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

Added common deserializer (#109)

Browse source
This commit is contained in:
German Osin 2020-11-02 15:21:07 +03:00 • committed by GitHub
parent e9a6b52f2a
commit 88cc301bb6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 49 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/deserialization/DeserializationService.java
View file

@ -1,25 +1,24 @@
package com.provectus.kafka.ui.cluster.deserialization;
import lombok.RequiredArgsConstructor;
import java.util.Map;
import java.util.stream.Collectors;
import javax.annotation.PostConstruct;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.provectus.kafka.ui.cluster.model.ClustersStorage;
import com.provectus.kafka.ui.cluster.model.KafkaCluster;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.Map;
import java.util.stream.Collectors;
@Component
@RequiredArgsConstructor
public class DeserializationService {
private final ClustersStorage clustersStorage;
private final ObjectMapper objectMapper;
private Map<String, RecordDeserializer> clusterDeserializers;
@PostConstruct
public void init() {
this.clusterDeserializers = clustersStorage.getKafkaClusters().stream()
@ -30,11 +29,7 @@ public class DeserializationService {
}
private RecordDeserializer createRecordDeserializerForCluster(KafkaCluster cluster) {
if (StringUtils.isEmpty(cluster.getSchemaRegistry())) {
return new SimpleRecordDeserializer();
} else {
return new SchemaRegistryRecordDeserializer(cluster);
}
return new SchemaRegistryRecordDeserializer(cluster, objectMapper);
}
public RecordDeserializer getRecordDeserializerForCluster(KafkaCluster cluster) {

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

@ -1,55 +1,56 @@
package com.provectus.kafka.ui.cluster.deserialization;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.utils.Bytes;
import io.confluent.kafka.schemaregistry.SchemaProvider;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.provectus.kafka.ui.cluster.model.KafkaCluster;
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.exceptions.RestClientException;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import lombok.extern.log4j.Log4j2;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.utils.Bytes;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.provectus.kafka.ui.cluster.model.KafkaCluster;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
@Log4j2
@RequiredArgsConstructor
public class SchemaRegistryRecordDeserializer implements RecordDeserializer {
private final static int CLIENT_IDENTITY_MAP_CAPACITY = 100;
private final KafkaCluster cluster;
private final SchemaRegistryClient schemaRegistryClient;
private KafkaAvroDeserializer avroDeserializer;
private ObjectMapper objectMapper;
private StringDeserializer stringDeserializer;
private final KafkaAvroDeserializer avroDeserializer;
private final ObjectMapper objectMapper;
private final StringDeserializer stringDeserializer;
private final Map<String, MessageFormat> topicFormatMap = new ConcurrentHashMap<>();
public SchemaRegistryRecordDeserializer(KafkaCluster cluster) {
public SchemaRegistryRecordDeserializer(KafkaCluster cluster, ObjectMapper objectMapper) {
this.cluster = cluster;
this.objectMapper = objectMapper;
List<String> endpoints = Collections.singletonList(cluster.getSchemaRegistry());
List<SchemaProvider> providers = Collections.singletonList(new AvroSchemaProvider());
this.schemaRegistryClient = new CachedSchemaRegistryClient(endpoints, CLIENT_IDENTITY_MAP_CAPACITY, providers, Collections.emptyMap());
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.avroDeserializer = new KafkaAvroDeserializer(schemaRegistryClient);
this.objectMapper = new ObjectMapper();
this.avroDeserializer = Optional.ofNullable(this.schemaRegistryClient)
.map(KafkaAvroDeserializer::new)
.orElse(null);
this.stringDeserializer = new StringDeserializer();
}
@ -83,12 +84,14 @@ public class SchemaRegistryRecordDeserializer implements RecordDeserializer {
private MessageFormat detectFormat(ConsumerRecord<Bytes, Bytes> record) {
String avroSchema = String.format(cluster.getSchemaNameTemplate(), record.topic());
if (schemaRegistryClient != null) {
try {
schemaRegistryClient.getAllVersions(avroSchema);
return MessageFormat.AVRO;
} catch (RestClientException | IOException e) {
log.info("Failed to get Avro schema for topic {}", record.topic());
}
}
try {
parseJsonRecord(record);
@ -102,7 +105,7 @@ public class SchemaRegistryRecordDeserializer implements RecordDeserializer {
private Object parseAvroRecord(ConsumerRecord<Bytes, Bytes> record) throws IOException {
String topic = record.topic();
if (record.value()!=null) {
if (record.value()!=null && avroDeserializer !=null) {
byte[] valueBytes = record.value().get();
GenericRecord avroRecord = (GenericRecord) avroDeserializer.deserialize(topic, valueBytes);
byte[] bytes = AvroSchemaUtils.toJson(avroRecord);