如何监控milvus-kafka的消息 求助！

[RamPageMz]

大佬们好，刚接触milvus没多久，现在集群上用helm部署了一套milvus，MQ用的是kafka，我现在想对milvus的kafka消息进行监听（用java程序），现在已经可以：

• 连接milvus的kafka
• 获取当前kafka的topic list

现在有两个问题

1. 这些topic的含义是什么？我现在看到的了topic格式有类似

• by-dev-rootcoord-dml-X
• document-process-request-X
• by-dev-replicate-msg
• document-process-response
  这几种topic处理的数据有什么不一样的含义吗

2. 我想用另一个消费组去消费这些topic的数据进行查看，但是出现乱码
   现在使用的consumer配置都是基础的默认的

Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group-0116-1");
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");


for (ConsumerRecord<byte[], byte[]> record : records) {
//    String cKey = new String(record.key(), StandardCharsets.UTF_8);
      String cValue = new String(record.value(), StandardCharsets.UTF_8);
      record.headers().forEach(header -> {
          System.out.println("Header: " + header.key()+", " + header.value() + " " + new String(header.value(), StandardCharsets.UTF_8));
      });
      System.out.println("Raw Key: " + record.key() + ", Value: " + record.value());
      System.out.println("New Key: " + record.key() + ", Value: " + cValue);
  }

然后获取到的数据string后输出，这样出现了很多乱码
有些乱码包含了一些我们db和collection的名称，有些是完全乱码

从Topic by-dev-rootcoord-dml_13 中读取数据：
Raw Key: null, Value: [B@d6e7bab
New Key: null, Value: 
���	������Ũ� ��


从Topic by-dev-replicate-msg 中读取数据：
Header: traceparent, [B@71def8f8 00-34507deb176634b9632e316e08886222-4a57065fc35ae4a5-00
Raw Key: null, Value: [B@f736069
New Key: null, Value: 
�����������˨��������˨� ����loadTest��testCollection1007

这个有什么办法反序列化么？ 以及这些message的格式结构在哪里可以看到定义

[yhmo]

milvus.yaml的msgChannel定义了所使用的各种topic的名称，从里面的注释可以看到分别是做什么用的

milvus/configs/milvus.yaml

Line 708 in 93ce4b1

msgChannel:

随着milvus内部的重构，有的topic可能已经弃用。以你看见的topic为准

milvus写入kafka/pulsar的消息是包装过的，比如insert请求的message是在这里面生成： https://github.com/milvus-io/milvus/blob/master/internal/proxy/task_insert.go 这里的insertTask.insertMsg就是要传入kafka的东西。

通过读取代码来获知具体的格式。这点我也帮不了你，因为我也是要读代码才知道

[RamPageMz]

如果我把某个topic的消费到 然后原封不动再发个这个topic （还是以byte array的方式） 是不是相当于会让milvus自动执行了插入数据等操作

[yhmo]

如果我把某个topic的消费到 然后原封不动再发个这个topic （还是以byte array的方式） 是不是相当于会让milvus自动执行了插入数据等操作

理论上是。
https://github.com/zilliztech/milvus-cdc 这个工具就是把上游的milvus的消息流中的数据取出转发给下游milvus的消息流。
如果把消息取出再次发给同一个milvus的消息流，对于insert的数据，就会出现重复的id和重复的时间戳，不确定会不会出妖鹅子。

[RamPageMz]

cdc应该是用来做两个集群的同步是吧，这种方案我考虑到是不是有这么几个不确定的事

• cdc服务稳定性，一旦服务中断，两个数据库产生了差异，是不是就得先停机做全量备份恢复 ，再开启新的cdc呀
• 相当于得有两个集群和两倍的资源消耗

[RamPageMz]

我还看到官网有和kafka和milvus的connector，这个挺好的，不过是不是一次只能给一个collection用啊 ，可以同时给多个collection 么

[yhmo]

设计上每次write只为一个collection灌数据。目前spark-milvus这个工具缺少人力维护