使用模式注册表API验证Kafka消息

Question

我正在实现一个生成Kafka消息的过程，每个消息都应该有由模式注册表验证的模式。为了开发，我使用docker运行kafka和模式注册表，我的模式由模式注册表ui进行注册。

看起来我的模式没有被验证或者我缺少一些配置。我的生产者类有以下代码：

package br.com.xx.realtime_transformation.producers;

import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;

import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.UUID;

public class KafkaEventProducer {

    private final String bootstrapServers;
    private final String schemaRegistryUrl;
    private final KafkaProducer<String, GenericRecord> kafkaProducer;

    public KafkaEventProducer(String bootstrapServers, String schemaRegistryUrl) {
        this.bootstrapServers = bootstrapServers;
        this.schemaRegistryUrl = schemaRegistryUrl;
        this.kafkaProducer = getProducer();
    }

    private KafkaProducer<String, GenericRecord> getProducer() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bootstrapServers);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
        props.put(AbstractKafkaAvroSerDeConfig.AUTO_REGISTER_SCHEMAS, false);
        props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, this.schemaRegistryUrl);

        return new KafkaProducer<>(props);
    }

    public void send(String topic, GenericRecord event) {
        try {
            String key = UUID.randomUUID().toString();
            final ProducerRecord producerRecord = new ProducerRecord<>(topic, key, event);

            this.kafkaProducer.send(producerRecord);
        } catch (final SerializationException e) {
            e.printStackTrace();
        }

    }

}

大多数时候，我会收到一个错误，如“Schema not found”，当没有引发此异常时，我的消息不会被验证，它只是将消息发送到另一个主题。

是否缺少任何类型的配置？

Answer 1

我不确定我是否理解这个问题，但您似乎在询问模式验证。

这将在5.4版本中处理

Answer 2

Kafka Producer使用模式注册表，而不使用模式注册表。

没有模式注册表-以下是没有模式注册表的Kafka生产者示例

https://www.devglan.com/apache-kafka/apache-kafka-java-example

使用架构注册表

schema.registry.url参数只是指向我们存储模式的位置。我们需要提供模式定义，如下所示。您没有提供模式定义，而是使用模式注册表，因此您遇到了问题。

{
  "namespace": "com.example",
  "type": "record",
  "name": "Employee",
  "doc" : "Represents an Employee at a company",
  "fields": [
      {"name": "firstName", "type": "string", "doc": "The persons given name"},
      {"name": "lastName", "type": "string"},
      {"name": "age",  "type": "int", "default": -1},
      {"name": "emails", "default":[], "type":{"type": "array", "items": "string"}},
      {"name": "phoneNumber",  "type": "string"}
   ]
} 

Schema Registry的工作流程如下所示。

您可以在以下URL上找到详细的解释。https://mapr.com/docs/61/Kafka/KafkaSchemaRegistry/KafkaSchemaRegistryDemo.html https://aseigneurin.github.io/2018/08/02/kafka-tutorial-4-avro-and-schema-registry.html https://www.confluent.jp/blog/kafka-connect-tutorial-transfer-avro-schemas-across-schema-registry-clusters/