StreamsDSLAndProcessorExample.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.kafka.streams;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.kstream.ValueTransformer;
import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;

import java.time.Instant;
import java.util.Properties;

public class StreamsDSLAndProcessorExample {

    public static void main(String[] args) {

        final Properties properties = new Properties();
        properties.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "test-application");
        properties.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        properties.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

        final String storeName = "stateStore";
        final StreamsBuilder builder = new StreamsBuilder();


        final StoreBuilder<KeyValueStore<String, String>> storeBuilder = Stores.keyValueStoreBuilder(
                                                                         Stores.persistentKeyValueStore(storeName),
                                                                         Serdes.String(),
                                                                         Serdes.String());

        builder.addStateStore(storeBuilder);


        builder.<String, String>stream("input")
               .filter((k,v) -> v.endsWith("FOO"))
               .transformValues(() -> new SimpleValueTransformer(storeName), storeName)
               .to("output");

        final Topology topology = builder.build(properties);
        System.out.println(topology.describe());

        final KafkaStreams streams = new KafkaStreams(topology, properties);
        streams.start();


    }


    static class SimpleValueTransformer implements ValueTransformerWithKey<String, String, String> {

        private String storeName;
        private KeyValueStore<String, String> store;

        public SimpleValueTransformer(String storeName) {
            this.storeName = storeName;
        }

        @Override
        public void init(final ProcessorContext context) {
             store = (KeyValueStore) context.getStateStore(storeName);
        }

        @Override
        public String transform(final String key, final String value) {
            String persistedValue = store.get(key);
            final String updatedValue = value + "_" + Instant.now().toString();

            if(persistedValue == null) {
                 persistedValue = updatedValue;
            }

            store.put(key, updatedValue);
            return persistedValue;
        }

        @Override
        public void close() {

        }
    }
}