diff --git a/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/HistoryServiceFlinkJob.java b/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/HistoryServiceFlinkJob.java index 6655b52ec3020f46bb8a37c7124ee870fa663573..0b732d5b4a8393275b5740da088aae5fa462b95f 100644 --- a/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/HistoryServiceFlinkJob.java +++ b/theodolite-benchmarks/uc1-flink/src/main/java/theodolite/uc1/application/HistoryServiceFlinkJob.java @@ -1,6 +1,7 @@ package theodolite.uc1.application; import org.apache.commons.configuration2.Configuration; +import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; @@ -61,7 +62,8 @@ public final class HistoryServiceFlinkJob { stream .rebalance() .map(new GsonMapper()) - .flatMap((record, c) -> LOGGER.info("Record: {}", record)); + .flatMap((record, c) -> LOGGER.info("Record: {}", record)) + .returns(Types.GENERIC(Object.class)); // Will never be used } /**