@@ -0,0 +1,2567 @@
+package com.codespair.mockstocks.service.kafka.spring.stream;%0A%0Aimport com.fasterxml.jackson.databind.JsonNode;%0Aimport lombok.extern.slf4j.Slf4j;%0Aimport org.apache.kafka.common.serialization.Deserializer;%0Aimport org.apache.kafka.common.serialization.Serde;%0Aimport org.apache.kafka.common.serialization.Serdes;%0Aimport org.apache.kafka.common.serialization.Serializer;%0Aimport org.apache.kafka.connect.json.JsonDeserializer;%0Aimport org.apache.kafka.connect.json.JsonSerializer;%0Aimport org.apache.kafka.streams.StreamsConfig;%0Aimport org.apache.kafka.streams.kstream.KStream;%0Aimport org.apache.kafka.streams.kstream.KStreamBuilder;%0Aimport org.apache.kafka.streams.kstream.KTable;%0Aimport org.apache.kafka.streams.processor.WallclockTimestampExtractor;%0Aimport org.springframework.context.annotation.Bean;%0Aimport org.springframework.context.annotation.Configuration;%0Aimport org.springframework.context.annotation.DependsOn;%0Aimport org.springframework.kafka.annotation.EnableKafka;%0Aimport org.springframework.kafka.annotation.EnableKafkaStreams;%0Aimport org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;%0A%0Aimport java.util.HashMap;%0Aimport java.util.Map;%0A%0A@Slf4j%0A@Configuration%0A@DependsOn(%22streamChain%22)%0A@EnableKafka%0A@EnableKafkaStreams%0Apublic class StreamConfiguration %7B%0A%0A    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)%0A    public StreamsConfig kStreamsConfigs() %7B%0A        log.info(%22loading spring kafka stream configuration%22);%0A        final Serializer%3CJsonNode%3E jsonSerializer = new JsonSerializer();%0A        final Deserializer%3CJsonNode%3E jsonDeserializer = new JsonDeserializer();%0A        final Serde%3CJsonNode%3E jsonSerde = Serdes.serdeFrom(jsonSerializer, jsonDeserializer);%0A%0A        Map%3CString, Object%3E props = new HashMap%3C%3E();%0A        props.put(StreamsConfig.APPLICATION_ID_CONFIG, %22spring-streams%22);%0A        props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());%0A        props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, jsonSerde);%0A        props.put(StreamsConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class.getName());%0A        return new StreamsConfig(props);%0A    %7D%0A%0A    @Bean%0A    public KStream%3CString, JsonNode%3E kStream() %7B%0A        KStreamBuilder kStreamBuilder = new KStreamBuilder();%0A        KStream%3CString, JsonNode%3E stream = kStreamBuilder.stream(%22amex-quotes%22);%0A        KTable%3CString, Long%3E countsBySymbol = stream.groupByKey().count(%22Counts%22);%0A        countsBySymbol.to(Serdes.String(), Serdes.Long(), %22spring-streams%22);%0A        return stream;%0A    %7D%0A%7D%0A

create spring stream configuration class.