Migrate Kafka Streams code to Spring Cloud Stream?

Question

Will the Spring cloud stream support the below Kafka Streams application. Below is code in excerpt of the Kafka sample app. Any feedback or support is appreciated.

        ...
        StreamsBuilder streamsBuilder = new StreamsBuilder();

        KStream<String, Purchase> purchaseKStream = streamsBuilder.stream.....
        KStream<String, PurchasePattern> patternKStream = purchaseKStream.mapValues...
        patternKStream.print(Printed.<String, PurchasePattern>toSysOut().withLabel("patterns"));
        patternKStream.to("patterns", Produced.with(stringSerde, purchasePatternSerde));
        purchaseKStream.print(Printed.<String, Purchase>toSysOut().withLabel("purchases"));
        purchaseKStream.to("purchases", Produced.with(stringSerde, purchaseSerde));

        // adding State to processor
        String rewardsStateStoreName = "rewardsPointsStore";
        RewardsStreamPartitioner streamPartitioner = new RewardsStreamPartitioner();
        KeyValueBytesStoreSupplier storeSupplier = Stores.inMemoryKeyValueStore(rewardsStateStoreName);
        StoreBuilder<KeyValueStore<String, Integer>> storeBuilder = Stores.keyValueStoreBuilder(storeSupplier.....
        streamsBuilder.addStateStore(storeBuilder);
        KStream<String, Purchase> transByCustomerStream = purchaseKStream.through("customer_transactions",....
        KStream<String, RewardAccumulator> statefulRewardAccumulator = transByCustomerStream
                .transformValues(() -> new PurchaseRewardTransformer(rewardsStateStoreName), rewardsStateStoreName);
        statefulRewardAccumulator.print(Printed.<String, RewardAccumulator>toSysOut().withLabel("stateful-rewards"));
        statefulRewardAccumulator.to("rewards", Produced.with(stringSerde, rewardAccumulatorSerde));

        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), getProperties());

sobychacko sobychacko · Accepted Answer · 2019-12-03T16:40:46

Assuming that you are using Spring Cloud Stream Horsham (3.0.0) release, the following pseudo-code should work. I didn't test this code, but this should work with the right configuration for the destination, etc. Please take a look at the docs for that.

@SpringBootApplicaiton
public class SampleApp {

  public static void main(String[] args) {
   SpringApplication.run(SampleApp.class, args);
  }

   @Bean
   public Function<KStream<String, Purchase>, KStream<String, RewardAccumulator>> process() {
     return purchaseKStream -> {
           purchaseKStream.print(Printed.<String, Purchase>toSysOut().withLabel("purchases"));
           KStream<String, PurchasePattern> patternKStream = purchaseKStream.mapValues...
           patternKStream.print(Printed.<String, PurchasePattern>toSysOut().withLabel("patterns"));
           patternKStream.to("patterns", Produced.with(stringSerde, purchasePatternSerde));
                purchaseKStream.to("purchases", Produced.with(stringSerde, purchaseSerde));

           KStream<String, Purchase> transByCustomerStream = purchaseKStream.through("customer_transactions",....
           KStream<String, RewardAccumulator> statefulRewardAccumulator = transByCustomerStream.transformValues(()
                statefulRewardAccumulator.print(Printed.<String, RewardAccumulator>toSysOut().withLabel("stateful-rewards"));

           return statefulRewardAccumulator;
       }        
   }

   @Bean
   public StoreBuilder storeBuilder() {
     String rewardsStateStoreName = "rewardsPointsStore";
     RewardsStreamPartitioner streamPartitioner = new RewardsStreamPartitioner();
     KeyValueBytesStoreSupplier storeSupplier = Stores.inMemoryKeyValueStore(rewardsStateStoreName);
     StoreBuilder<KeyValueStore<String, Integer>> storeBuilder = Stores.keyValueStoreBuilder(storeSupplier.....
     return storeBuilder;
   }

}

Migrate Kafka Streams code to Spring Cloud Stream?

1 Answers