building a GlobalKTable

[pitermarx]

I want to build a global ktable from a stream.
The way I'm doing right now is by aggregating the stream into a KTable, and then using the "changelog" topic as the source for the GlobalKTable.

var config = new StreamConfig
{
    ApplicationId = "test",
    BootstrapServers = "localhost:9092",
    SchemaRegistryUrl = "http://localhost:8081",
    AutoRegisterSchemas = true,
    DefaultKeySerDes = new StringSerDes(),
    DefaultValueSerDes = new AvroSerdes<ComplexType>()
};

StreamBuilder builder = new();

IKTable<string, ComplexType> ktable = builder
    .Stream<string, ComplexType>("inputTopic")
    .GroupByKey()
    .Aggregate<ComplexType>(
        () => null!,
        // let's imagine a complex aggregation
        (key, value, aggregation) => value,
        // I really dont need this in memory, can I skip this part?
        InMemory.As<string, ComplexType>("storename"));

IGlobalKTable<string, ComplexType> gkTable = builder
    .GlobalTable<string, ComplexType>(
        // do I need to "guess" the topic name? is there any other way?
        $"{config.ApplicationId}-storename-changelog");

record ComplexType(string some, string values);

The guessing part, doesn't feel right to me, but I can't find any help on this online.
Anyone ever did something like this? Taking a topic and then aggregating it into a global KTable?
I could do a .ToStream().To() a known topic, but wouldn't that duplicate the data in Kafka?

[LGouellec]

Hi @pitermarx,

Unfortunately, you can't consume a topic and built a global store manually (you can do it in Kafka Streams JAVA).
With Streamiz, you have to consume your topic, aggregate, forward the result of your aggregation into an another topic and use this topic as a globalKtable.

It's not recommanded to use the changelog topic because it's an internal topic and we don't manage it all.

So regarding your use case, it could look like this :

StreamBuilder builder = new();

IKTable<string, ComplexType> ktable = builder
    .Stream<string, ComplexType>("inputTopic")
    .GroupByKey()
    .Aggregate<ComplexType>(
        () => null!,
        // let's imagine a complex aggregation
        (key, value, aggregation) => value)
    .ToStream()
    .To("output");

IGlobalKTable<string, ComplexType> gkTable = builder.GlobalTable<string, ComplexType>("output");

Best regards,

[pitermarx]

thks for the quick reply
I also sugested doing that on my question. But doesn't that lead to a duplication of data?
Once in the changelog topic, and another time in the "output" topic.

do you foresee any way of doing it in streamiz? is it intrinsically impossible or just not implemented?

If you have it in hand can you point me to an example in JAVA?

[LGouellec]

It not implemented for now.

In JAVA you can create your custom global store from a specific topic like :

builder.addGlobalStore(
                Stores.keyValueStoreBuilder(
                        Stores.persistentKeyValueStore("global-store"), Serdes.String(), Serdes.String()),
                "topic",
                Consumed.with(Serdes.String(), Serdes.String()),
                () -> new GlobalStoreUpdater("global-store"));

    private static class GlobalStoreUpdater<K, V> implements Processor<K, V, Void, Void> {
        private final String storeName;

        public GlobalStoreUpdater(final String storeName) {
            this.storeName = storeName;
        }

        private KeyValueStore<K, V> store;

        @Override
        public void init(final ProcessorContext<Void, Void> processorContext) {
            store = processorContext.getStateStore(storeName);
        }

        @Override
        public void process(final Record<K, V> record) {
            // We are only supposed to put operation the keep the store updated.
            // We should not filter record or modify the key or value
            // Doing so would break fault-tolerance.
            // see https://issues.apache.org/jira/browse/KAFKA-7663
            store.put(record.key(), record.value());
        }

        @Override
        public void close() {
            // No-op
        }
    }

But be carefuk-ll, it's highly recommanded to not change the key/value inside the processor. Why ? Because during the restoration phase, the processor is not called, so you can restore a state store differently than the real time processing.
https://issues.apache.org/jira/browse/KAFKA-7663

If you want to create a global store from a topic and change the key and/or the value, you have to Stream-ChangeKey-To + create GlobalKTable from the output topic even with Kafka Streams JAVA

[pitermarx]

thanks for the support