用于 Schema 注册的 JavaScript 客户端^{TM 电话}

Conluent 的 JavaScript 客户端 Schema 书记官处 支持Avro,Protobuf和JSON Schema,并旨在与 Confluent 用于 Apache Kafka 的 JavaScript 客户端。 。 。 . 目的是按照Schema注册公司的其他客户提供高性能、可靠和易于使用的JavaScript客户端 就像我们 走开, (中文(简体) ), . NET 网络, (中文(简体) ), 财务报告和审定财务报表 爪哇语 客户.

安装

npm install @confluentinc/schemaregistry

开始

下面是使用Avro系列化的简单例子,与Schema注册客户端和KafkaJS客户端.

const { Kafka } = require('@confluentinc/kafka-javascript').KafkaJS;
const { SchemaRegistryClient, SerdeType, AvroSerializer, AvroDeserializer} = require('@confluentinc/schemaregistry');

const registry = new SchemaRegistryClient({ baseURLs: ['http://localhost:8081'] })
const kafka = new Kafka({
  kafkaJS: {
    brokers: ['localhost:9092']
  }
});

let consumer = kafka.consumer({
  kafkaJS: {
    groupId: "test-group",
    fromBeginning: true,
  },
});
let producer = kafka.producer();

const schema = {
  type: 'record',
  namespace: 'examples',
  name: 'RandomTest',
  fields: [
    { name: 'fullName', type: 'string' }
  ],
};

const topicName = 'test-topic';
const subjectName = topicName + '-value';

const run = async () => {
  // Register schema
  const id = await registry.register(
    subjectName,
    {
      schemaType: 'AVRO',
      schema: JSON.stringify(schema)
    }
  )

  // Create an Avro serializer
  const ser = new AvroSerializer(registry, SerdeType.VALUE, { useLatestVersion: true });

  // Produce a message with the schema
  await producer.connect()
  const outgoingMessage = {
    key: 'key',
    value: await ser.serialize(topicName, { fullName: 'John Doe' }),
  }
  await producer.send({
    topic: topicName,
    messages: [outgoingMessage]
  });
  console.log("Producer sent its message.")
  await producer.disconnect();
  producer = null;

  // Create an Avro deserializer
  const deser = new AvroDeserializer(registry, SerdeType.VALUE, {});

  await consumer.connect()
  await consumer.subscribe({ topic: topicName })

  let messageRcvd = false;
  await consumer.run({
    eachMessage: async ({ message }) => {
      const decodedMessage = {
        ...message,
        value: await deser.deserialize(topicName, message.value)
      };
      console.log("Consumer received message.\nBefore decoding: " + JSON.stringify(message) + "\nAfter decoding: " + JSON.stringify(decodedMessage));
      messageRcvd = true;
    },
  });

  // Wait around until we get a message, and then disconnect.
  while (!messageRcvd) {
    await new Promise((resolve) => setTimeout(resolve, 100));
  }

  await consumer.disconnect();
  consumer = null;
}

run().catch (async e => {
  console.error(e);
  consumer && await consumer.disconnect();
  producer && await producer.disconnect();
  process.exit(1);
})

特征和限制

• Avro、Protobuf和JSON Schema的支持
• 支持 CSFLE( 连续的字段加密)
• 支持数据质量规则
• 支持为Avro和JSON Schema制定移民规则
• 对 OAuth 的支持

捐款

错误报告和反馈以"Github Issues"的形式获得赞赏. 关于捐款准则,请参见: 捐款额