我有一个代码，实现了kafkastream在springboot应用程序中的工作：

public Consumer<KStream<String, ContentModel>> stream() {
    return in -> in.groupByKey(Grouped.with(Serdes.String(), new JsonSerde<>(ContentModel.class)))      
}

private ContentModel mergeObjects(final ContentModel reducer, final ContentModel materialized) {
    return reducer;
}

问题是我的应用程序在其他情况下使用自己的objectmapper bean（例如，读取包含json内容的文件）。由于jsonserge还包含“内部”自己的objectmapper，所以2个objectmapper的示例会引起“混乱”。
我想办法把两个物体Map器放在两边。
为此，我尝试定制用于kafkastream处理的jsonserge：
在配置中：

@Bean
public JsonSerde<ContentModel> jsonSerde() {
    ObjectMapper kafkaObjectMapper = new ObjectMapper();
    return new JsonSerde<>(ContentModel.class, kafkaObjectMapper);
}

上课时间：

@Autowired
private JsonSerde<ContentModel> jsonSerde;

public Consumer<KStream<String, ContentModel>> stream() {
    return in -> in.groupByKey(Grouped.with(Serdes.String(), jsonSerde))
            .reduce(this::mergeObjects, Materialized.as(readMessagesStore.getStoreName()));
}

但是，我收到以下错误：

java.lang.IllegalStateException: KafkaStreams is not running. State is ERROR.
at org.apache.kafka.streams.KafkaStreams.validateIsRunning(KafkaStreams.java:294) ~[kafka-streams-2.3.1.jar!/:?]
at org.apache.kafka.streams.KafkaStreams.store(KafkaStreams.java:1066) ~[kafka-streams-2.3.1.jar!/:?]

我做错什么了？