java - Spring boot TopicRecordNameStrategy 使用量消費者がそれに関連する recordType を読み取る

Question

2 つの異なる API を介してイベントを kafka に発行する例を試しています。1 つは作成イベントを取得し、もう 1 つは同じトピック ユーザーの更新イベントを取得します。

これらの API はどちらも、異なるスキーマを持つ同じ kafka トピックにメッセージを発行します。

基本的に UserCreated、UserUpdated スキーマがユーザー トピックに公開されます。

コンシューマーは UserUpdated イベントのみに関心があるため、TopicRecordNameStrategy をサポートするコンシューマーをどのようにセットアップできるか疑問に思っています。現在、ユーザーのトピックをリッスンしている 2 つのコンシューマーがいて、両方ともメッセージを受信して​​います。

プロデューサーとコンシューマーの両方が同じサービスにあるコンフルエントなカフカに接続するサンプル スプリング ブート アプリを作成しました

作成と更新のための 2 つのエンドポイントをサポートする REST API があります。この残りの API はプロデューサーにメッセージを発行します

これが私のコードです。

アプリケーションyml

  name: users
  partitions-num: 3
  replication-factor: 3
server:
  port: 9080
spring:
  kafka:
    properties:
      bootstrap.servers: *****
      value.subject.name.strategy: io.confluent.kafka.serializers.subject.TopicRecordNameStrategy
      sasl:
        jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username='****' password='****';
        mechanism: PLAIN
      security.protocol: SASL_SSL

      # CCloud Schema Registry Connection parameter
      schema.registry.url: ******
      basic.auth.credentials.source: USER_INFO
      schema.registry.basic.auth.user.info: *****:*****
    consumer:
      group-id: local-test-user-consumergroup
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
    template:
      default-topic:
logging:
  level:
    root: info

RestController.java

import com.aligntech.UserUpdated;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping(value = "/user")
public class KafkaController {

    private final Producer producer;

    @Autowired
    KafkaController(Producer producer) {
        this.producer = producer;
    }

    @PostMapping(value = "/create")
    public void create(@RequestParam("name") String name, @RequestParam("age") Integer age) {
        UserCreated build = UserCreated.newBuilder().setAge(age).setName(name).build();
        this.producer.sendMessage(build);
    }


    @PutMapping(value = "/update")
    public void update(@RequestParam("name") String name, @RequestParam("age") Integer age) {
        UserUpdated build = UserUpdated.newBuilder().setAge(age).setName(name).setUpdated(true).build();
        this.producer.sendMessage(build);
    }
}

Producer.java

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class Producer {


    private final KafkaTemplate<String, User> kafkaTemplate;

    @Value("${topic.name}")
    private String topic;

    @Autowired
    public Producer(KafkaTemplate<String, User> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    void sendMessage(com.aligntech.UserCreated user) {
        ProducerRecord record = new ProducerRecord<>(topic, user);
        kafkaTemplate.send(record);
    }


    void sendMessage(com.aligntech.UserUpdated user) {
        ProducerRecord record = new ProducerRecord<>(topic, user);
        kafkaTemplate.send(record);
    }
}

コンシューマ.java

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class Consumer {

    @KafkaListener(topics = "users", groupId = "user-updated-group")
    public void consumeUserUpdated(ConsumerRecord<String, com.aligntech.UserUpdated> record) {
        log.info("Read UserUpdated Record");
        log.info(record.topic() + ":" + record.key() + ":" + record.headers());
        log.info(String.format("Consumed message -> %s", record.value()));
    }


    @KafkaListener(topics = "users", groupId = "user-created-group")
    public void consumeUserCreated(ConsumerRecord<String, com.aligntech.UserCreated> record) {

        log.info("Read UserCreated Record");
        log.info(record.topic() + ":" + record.key() + ":" + record.headers());
        log.info(String.format("Consumed message -> %s", record.value()));
    }
}

出力

2021-05-24 14:51:54.220  INFO 3878 --- [ntainer#1-0-C-1] com.****.kafka.kafkatest.Consumer  : Read UserCreated Record
2021-05-24 14:51:54.223  INFO 3878 --- [ntainer#1-0-C-1] com.****.kafka.kafkatest.Consumer  : users:null:RecordHeaders(headers = [], isReadOnly = false)
2021-05-24 14:51:54.223  INFO 3878 --- [ntainer#1-0-C-1] com.****.kafka.kafkatest.Consumer  : Consumed message -> {"name": "UserUpdated", "age": 25, "updated": true}
2021-05-24 14:51:54.508  INFO 3878 --- [ntainer#0-0-C-1] com.****.kafka.kafkatest.Consumer  : Read UserUpdated Record
2021-05-24 14:51:54.508  INFO 3878 --- [ntainer#0-0-C-1] com.****.kafka.kafkatest.Consumer  : users:null:RecordHeaders(headers = [], isReadOnly = false)
2021-05-24 14:51:54.508  INFO 3878 --- [ntainer#0-0-C-1] com.****.kafka.kafkatest.Consumer  : Consumed message -> {"name": "UserUpdated", "age": 25, "updated": true}