ftrossbach

Vertx.clusteredVertx(new VertxOptions().setClusterManager(new HazelcastClusterManager()), handler -> {
   if(handler.succeeded()){
       Vertx vertx = handler.result();
    });

ftrossbach

{  
   "hostInfo":{  
      "host":"1.2.3.4",
      "port":42
   },
   "stateStoreNames":[  
      "totalVisitCount",
      "hourlyVisitCount",
      "sessionVisitCount"
   ],

ftrossbach

KStreamBuilder builder = new KStreamBuilder();
KStream<String, Long> visitsStream = builder.stream(Serdes.String(), Serdes.Long(), "visitsTopic");
KGroupedStream<String, Long> groupedStream = visitsStream.groupByKey();
KTable<String, Long> totalCount = groupedStream.count("totalVisitCount");
KTable<Windowed<String>, Long> windowedCount = groupedStream.count(TimeWindows.of(60 * 60 * 1000), "hourlyVisitCount");
groupedStream.count(SessionWindows.with(60 * 1000), "sessionVisitCount");