From 03b9bfa2809bddb0f8845502ec2b9e7fbd7219c2 Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Mon, 31 Aug 2020 23:31:37 +0530 Subject: [PATCH] Fix http-clicks-and-regions source --- .../http-clicks-and-region-ingest/pom.xml | 4 ---- .../samples/source/HttpClicksAndRegionIngest.java | 10 +++++++--- .../processor/UserClicksAndUserRegionsProcessor.java | 3 ++- 3 files changed, 9 insertions(+), 8 deletions(-) diff --git a/multi-io-samples/kafka-streams/user-clicks-per-region/http-clicks-and-region-ingest/pom.xml b/multi-io-samples/kafka-streams/user-clicks-per-region/http-clicks-and-region-ingest/pom.xml index e700ae7..82dc07e 100644 --- a/multi-io-samples/kafka-streams/user-clicks-per-region/http-clicks-and-region-ingest/pom.xml +++ b/multi-io-samples/kafka-streams/user-clicks-per-region/http-clicks-and-region-ingest/pom.xml @@ -89,10 +89,6 @@ prometheus-rsocket-spring 0.9.0 - - io.micrometer - micrometer-registry-datadog - io.pivotal.cfenv java-cfenv-boot diff --git a/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 b/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 index 73e8f22..b6fff01 100644 --- a/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 +++ b/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> clicksAndRegionsFlow() { - return IntegrationFlows.from( + Publisher> 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 toUserClicks(Message 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 toUserRegion(Message 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) { diff --git a/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 b/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 index 90e68a3..e09bfa8 100644 --- a/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 +++ b/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(); }