From 87f004366fa74e32e7a96aaf41a05c07f69433b7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?S=C3=B6ren=20Henning?= <soeren.henning@email.uni-kiel.de>
Date: Wed, 28 Apr 2021 17:10:48 +0200
Subject: [PATCH] Fix type info bug in UC1
---
.../theodolite/uc1/application/HistoryServiceFlinkJob.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
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 6655b52ec..0b732d5b4 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
}
/**
--
GitLab