Fix http-clicks-and-regions source

multi-io-samples/kafka-streams/user-clicks-per-region/http-clicks-and-region-ingest/pom.xml

@@ -89,10 +89,6 @@
       <artifactId>prometheus-rsocket-spring</artifactId>
       <version>0.9.0</version>
     </dependency>
-    <dependency>
-      <groupId>io.micrometer</groupId>
-      <artifactId>micrometer-registry-datadog</artifactId>
-    </dependency>
     <dependency>
       <groupId>io.pivotal.cfenv</groupId>
       <artifactId>java-cfenv-boot</artifactId>

multi-io-samples/kafka-streams/user-clicks-per-region/http-clicks-and-region-ingest/src/main/java/org/springframework/cloud/dataflow/samples/source/HttpClicksAndRegionIngest.java

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.dataflow.samples.source;
 
+import java.util.Collections;
 import java.util.function.Supplier;
 
 import org.reactivestreams.Publisher;
@@ -27,6 +28,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.MessageChannels;
 import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.http.support.DefaultHttpHeaderMapper;
 import org.springframework.integration.mapping.HeaderMapper;
@@ -43,13 +45,15 @@ public class HttpClicksAndRegionIngest {
 
 	@Bean
 	public Publisher<Message<byte[]>> clicksAndRegionsFlow() {
-		return IntegrationFlows.from(
+		Publisher<Message<byte[]>> publisher =  IntegrationFlows.from(
 				WebFlux.inboundChannelAdapter("/clicks", "/regions")
 						.requestPayloadType(byte[].class)
 						.statusCodeExpression(new ValueExpression<>(HttpStatus.ACCEPTED))
 						.mappedRequestHeaders("username")
 						.autoStartup(false))
+				.channel(MessageChannels.flux())
 				.toReactivePublisher();
+		return publisher;
 	}
 
 	@Bean
@@ -80,11 +84,11 @@ public class HttpClicksAndRegionIngest {
 	}
 
 	public Message<Long> toUserClicks(Message<byte[]> message) {
-		return new MutableMessage<>(Long.valueOf(new String(message.getPayload())), message.getHeaders());
+		return new MutableMessage<>(Long.valueOf(new String(message.getPayload())), Collections.singletonMap("username", message.getHeaders().get("username")));
 	}
 
 	public Message<String> toUserRegion(Message<byte[]> message) {
-		return new MutableMessage<>(new String(message.getPayload()), message.getHeaders());
+		return new MutableMessage<>(new String(message.getPayload()), Collections.singletonMap("username", message.getHeaders().get("username")));
 	}
 
 	public static void main(String[] args) {

multi-io-samples/kafka-streams/user-clicks-per-region/user-clicks-per-region-processor/src/main/java/org/springframework/cloud/dataflow/samples/processor/UserClicksAndUserRegionsProcessor.java

@@ -20,6 +20,7 @@ import java.util.function.BiFunction;
 
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.Grouped;
 import org.apache.kafka.streams.kstream.Joined;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
@@ -44,7 +45,7 @@ public class UserClicksAndUserRegionsProcessor {
 						(clicks, region) -> new RegionWithClicks(region == null ? "UNKNOWN" : region, clicks),
 						Joined.with(Serdes.String(), Serdes.Long(), null))
 				.map((user, regionWithClicks) -> new KeyValue<>(regionWithClicks.getRegion(), regionWithClicks.getClicks()))
-				.groupByKey(Serialized.with(Serdes.String(), Serdes.Long()))
+				.groupByKey(Grouped.with(Serdes.String(), Serdes.Long()))
 				.reduce((firstClicks, secondClicks) -> firstClicks + secondClicks)
 				.toStream();
 	}