Ktable to KGroupTable-架构不可用（未注册状态存储更改日志架构）

Question

我有一个Kafka主题-让我们活动-每日-聚合，我想使用KGroupTable进行聚合（添加/子）。所以我使用

final KTable<String, GenericRecord> inputKTable =
      builder.table("activity-daily-aggregate",Consumed.with(new StringSerde(), getConsumerSerde());

Note: getConsumerSerde - returns >> new GenericAvroSerde(mockSchemaRegistryClient)

2.Next Step,

inputKTable.groupBy(
        (key,value)->KeyValue.pair(KeyMapper.generateGroupKey(value), new JsonValueMapper().apply(value)),
        Grouped.with(AppSerdes.String(), AppSerdes.jsonNode())
        );

在第1步和第2步之前，我已经将MockSchemaRegistryClient配置为

 mockSchemaRegistryClient.register("activity-daily-aggregate-key",
                    Schema.parse(AppUtils.class.getResourceAsStream("/avro/key.avsc")));
       mockSchemaRegistryClient.register("activity-daily-aggregate-value",
                    Schema.parse(AppUtils.class.getResourceAsStream("/avro/daily-activity-aggregate.avsc")))

当我使用测试用例运行拓扑时，我在第2步得到一个错误。

组织。阿帕奇。Kafka。溪流。错误。StreamsException:进程中捕获异常。taskId=0_0，processor=KSTREAM-SOURCE-00000000 11，topic=activity daily aggregate，partition=0，offset=0，stacktrace=org。阿帕奇。Kafka。常见的错误。SerializationException:检索Avro架构时出错：{“type”：“record”，“name”：“FactActivity”，“namespace”：“com.ascendlearning.Avro”，“fields”：……}

原因：io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientExctive：架构未找到；错误代码：404001

当我向MockSchemaRegistrCyclient注册模式时，错误就会发生，

stream-app-id-activity-daily-aggregate-STATE-STORE-00000000 10变更日志密钥

stream-app-id-activity-daily-aggregate-STATE-STORE-00000000 10变更记录值=

我们需要做这一步吗？我想它可能会被拓扑自动处理

Answer 1

在博客上，https://blog.jdriven.com/2019/12/kafka-streams-topologytestdriver-with-avro/

当您在传递到TopologyTestDriver的两个属性中配置相同的mock://URL时，对于传递到createInputTopic和createOutputTopic的（反）序列化程序实例，所有（反）序列化程序都将使用相同的MockSchemaRegistryClient，并具有单个内存模式存储。

 // Configure Serdes to use the same mock schema registry URL
    Map<String, String> config = Map.of(
        AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, MOCK_SCHEMA_REGISTRY_URL);
    avroUserSerde.configure(config, false);
    avroColorSerde.configure(config, false);

    // Define input and output topics to use in tests
    usersTopic = testDriver.createInputTopic(
        "users-topic",
        stringSerde.serializer(),
        avroUserSerde.serializer());
    colorsTopic = testDriver.createOutputTopic(
        "colors-topic",
        stringSerde.deserializer(),
        avroColorSerde.deserializer());

在传递给输入/输出主题的serdes中，我没有传递模拟注册表客户机模式URL。