如何流到全局kafka表

我有一个Kafka Streams应用程序,它需要针对全局表加入一个传入流,然后经过一些处理,将聚合的结果写回该表:

KeyValueBytesStoreSupplier supplier = Stores.persistentKeyValueStore(
    storeName
);

Materialized<String, String, KeyValueStore<Bytes, byte[]>> m = Materialized.as(
    supplier
);

GlobalKTable<String, String> table = builder.globalTable(
    topic, m.withKeySerde(
        Serdes.String()
    ).withValueSerde(
        Serdes.String()
    )
);

stream.leftJoin(
    table
    ...
).groupByKey().aggregate(
    ... 
).toStream().through(
    topic, Produced.with(Serdes.String(), Serdes.String())
);

但是,当我尝试流入ktable changelog时,我得到以下错误: Invalid topology: Topic 'topic' has already been registered by another source.

如果我尝试聚合到存储本身,我会得到以下错误: InvalidStateStoreException: Store 'store' is currently closed .

如何对表进行联接并将其写回变更日志?

如果这是不可能的,一个涉及根据存储筛选传入日志的解决方案也可以工作。