You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2016/03/04 19:11:06 UTC
[11/50] [abbrv] incubator-beam git commit: [cleanup] various small
improvements
[cleanup] various small improvements
- removed duplicate declarations in pom.xml
- removed reference to junit.framework.*
- removed 'static' from interface declarations
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/602d8fe9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/602d8fe9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/602d8fe9
Branch: refs/heads/master
Commit: 602d8fe9b69dba2300a573559e7b60e286f116d1
Parents: 70ae13c
Author: smarthi <sm...@apache.org>
Authored: Thu Feb 11 17:33:51 2016 -0500
Committer: Davor Bonaci <da...@users.noreply.github.com>
Committed: Fri Mar 4 10:04:23 2016 -0800
----------------------------------------------------------------------
runners/flink/pom.xml | 5 -----
.../flink/dataflow/FlinkJobExecutionEnvironment.java | 3 +--
.../flink/dataflow/examples/streaming/AutoComplete.java | 6 ++----
.../flink/dataflow/examples/streaming/JoinExamples.java | 6 +++---
.../examples/streaming/KafkaWindowedWordCountExample.java | 2 +-
.../flink/dataflow/examples/streaming/WindowedWordCount.java | 3 +--
.../dataflow/translation/FlinkBatchPipelineTranslator.java | 3 ---
.../dataflow/translation/FlinkBatchTransformTranslators.java | 2 +-
.../translation/FlinkStreamingPipelineTranslator.java | 4 ----
.../translation/FlinkStreamingTransformTranslators.java | 5 ++---
.../translation/FlinkStreamingTranslationContext.java | 6 ++----
.../functions/FlinkCoGroupKeyedListAggregator.java | 2 +-
.../flink/dataflow/translation/wrappers/SinkOutputFormat.java | 2 --
.../dataflow/translation/wrappers/SourceInputFormat.java | 7 +++----
.../java/com/dataartisans/flink/dataflow/WriteSinkITCase.java | 2 +-
.../com/dataartisans/flink/dataflow/util/JoinExamples.java | 2 +-
16 files changed, 19 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/pom.xml
----------------------------------------------------------------------
diff --git a/runners/flink/pom.xml b/runners/flink/pom.xml
index 14693b8..cb784a0 100644
--- a/runners/flink/pom.xml
+++ b/runners/flink/pom.xml
@@ -73,11 +73,6 @@
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>${flink.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java</artifactId>
- <version>${flink.version}</version>
<scope>test</scope>
<type>test-jar</type>
</dependency>
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/FlinkJobExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/FlinkJobExecutionEnvironment.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/FlinkJobExecutionEnvironment.java
index 66d60fa..91b2f64 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/FlinkJobExecutionEnvironment.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/FlinkJobExecutionEnvironment.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.java.CollectionEnvironment;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.streaming.api.TimeCharacteristic;
-import org.apache.flink.streaming.api.environment.RemoteStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -224,7 +223,7 @@ public class FlinkJobExecutionEnvironment {
this.flinkStreamEnv.getConfig().setExecutionRetryDelay(3000);
}
- private final void checkInitializationState() {
+ private void checkInitializationState() {
if (this.options == null) {
throw new IllegalStateException("FlinkJobExecutionEnvironment is not initialized yet.");
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/AutoComplete.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/AutoComplete.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/AutoComplete.java
index 0245a7b..711d9fb 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/AutoComplete.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/AutoComplete.java
@@ -36,8 +36,6 @@ import org.joda.time.Duration;
import java.io.IOException;
import java.util.List;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
/**
* To run the example, first open a socket on a terminal by executing the command:
@@ -242,7 +240,7 @@ public class AutoComplete {
public void processElement(ProcessContext c) {
String word = c.element().value;
for (int i = minPrefix; i <= Math.min(word.length(), maxPrefix); i++) {
- KV kv = KV.of(word.substring(0, i), c.element());
+ KV<String, CompletionCandidate> kv = KV.of(word.substring(0, i), c.element());
c.output(kv);
}
}
@@ -349,7 +347,7 @@ public class AutoComplete {
*
* <p> Inherits standard Dataflow configuration options.
*/
- private static interface Options extends WindowedWordCount.StreamingWordCountOptions {
+ private interface Options extends WindowedWordCount.StreamingWordCountOptions {
@Description("Whether to use the recursive algorithm")
@Default.Boolean(true)
Boolean getRecursive();
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/JoinExamples.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/JoinExamples.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/JoinExamples.java
index b0cc4fa..9a5db64 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/JoinExamples.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/JoinExamples.java
@@ -53,8 +53,8 @@ public class JoinExamples {
static PCollection<String> joinEvents(PCollection<String> streamA,
PCollection<String> streamB) throws Exception {
- final TupleTag<String> firstInfoTag = new TupleTag<String>();
- final TupleTag<String> secondInfoTag = new TupleTag<String>();
+ final TupleTag<String> firstInfoTag = new TupleTag<>();
+ final TupleTag<String> secondInfoTag = new TupleTag<>();
// transform both input collections to tuple collections, where the keys are country
// codes in both cases.
@@ -118,7 +118,7 @@ public class JoinExamples {
}
}
- private static interface Options extends WindowedWordCount.StreamingWordCountOptions {
+ private interface Options extends WindowedWordCount.StreamingWordCountOptions {
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/KafkaWindowedWordCountExample.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/KafkaWindowedWordCountExample.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/KafkaWindowedWordCountExample.java
index 46c9bd6..42d3d88 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/KafkaWindowedWordCountExample.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/KafkaWindowedWordCountExample.java
@@ -72,7 +72,7 @@ public class KafkaWindowedWordCountExample {
}
}
- public static interface KafkaStreamingWordCountOptions extends WindowedWordCount.StreamingWordCountOptions {
+ public interface KafkaStreamingWordCountOptions extends WindowedWordCount.StreamingWordCountOptions {
@Description("The Kafka topic to read from")
@Default.String(KAFKA_TOPIC)
String getKafkaTopic();
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/WindowedWordCount.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/WindowedWordCount.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/WindowedWordCount.java
index 1d4a44b..b539245 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/WindowedWordCount.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/examples/streaming/WindowedWordCount.java
@@ -28,7 +28,6 @@ import com.google.cloud.dataflow.sdk.values.KV;
import com.google.cloud.dataflow.sdk.values.PCollection;
import org.joda.time.Duration;
-import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -81,7 +80,7 @@ public class WindowedWordCount {
}
}
- public static interface StreamingWordCountOptions extends com.dataartisans.flink.dataflow.examples.WordCount.Options {
+ public interface StreamingWordCountOptions extends com.dataartisans.flink.dataflow.examples.WordCount.Options {
@Description("Sliding window duration, in seconds")
@Default.Long(WINDOW_SIZE)
Long getWindowSize();
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkBatchPipelineTranslator.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkBatchPipelineTranslator.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkBatchPipelineTranslator.java
index 8c0183e..a1e4410 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkBatchPipelineTranslator.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkBatchPipelineTranslator.java
@@ -115,9 +115,6 @@ public class FlinkBatchPipelineTranslator extends FlinkPipelineTranslator {
}
private <T extends PTransform<?, ?>> void applyBatchTransform(PTransform<?, ?> transform, TransformTreeNode node, BatchTransformTranslator<?> translator) {
- if (this.batchContext == null) {
- throw new IllegalStateException("The FlinkPipelineTranslator is not yet initialized.");
- }
@SuppressWarnings("unchecked")
T typedTransform = (T) transform;
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkBatchTransformTranslators.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkBatchTransformTranslators.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkBatchTransformTranslators.java
index 8f64730..9a43d05 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkBatchTransformTranslators.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkBatchTransformTranslators.java
@@ -283,7 +283,7 @@ public class FlinkBatchTransformTranslators {
private static class ConsoleIOWriteTranslatorBatch implements FlinkBatchPipelineTranslator.BatchTransformTranslator<ConsoleIO.Write.Bound> {
@Override
public void translateNode(ConsoleIO.Write.Bound transform, FlinkBatchTranslationContext context) {
- PValue input = (PValue) context.getInput(transform);
+ PValue input = context.getInput(transform);
DataSet<?> inputDataSet = context.getInputDataSet(input);
inputDataSet.printOnTaskManager(transform.getName());
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingPipelineTranslator.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingPipelineTranslator.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingPipelineTranslator.java
index c8760c7..a8f4226 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingPipelineTranslator.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingPipelineTranslator.java
@@ -19,7 +19,6 @@ import com.google.cloud.dataflow.sdk.options.PipelineOptions;
import com.google.cloud.dataflow.sdk.runners.TransformTreeNode;
import com.google.cloud.dataflow.sdk.transforms.AppliedPTransform;
import com.google.cloud.dataflow.sdk.transforms.PTransform;
-import com.google.cloud.dataflow.sdk.transforms.join.CoGroupByKey;
import com.google.cloud.dataflow.sdk.values.PValue;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -101,9 +100,6 @@ public class FlinkStreamingPipelineTranslator extends FlinkPipelineTranslator {
}
private <T extends PTransform<?, ?>> void applyStreamingTransform(PTransform<?, ?> transform, TransformTreeNode node, StreamTransformTranslator<?> translator) {
- if (this.streamingContext == null) {
- throw new IllegalStateException("The FlinkPipelineTranslator is not yet initialized.");
- }
@SuppressWarnings("unchecked")
T typedTransform = (T) transform;
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingTransformTranslators.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingTransformTranslators.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingTransformTranslators.java
index 4c8cd4b..17583cd 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingTransformTranslators.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingTransformTranslators.java
@@ -76,8 +76,7 @@ public class FlinkStreamingTransformTranslators {
}
public static FlinkStreamingPipelineTranslator.StreamTransformTranslator<?> getTranslator(PTransform<?, ?> transform) {
- FlinkStreamingPipelineTranslator.StreamTransformTranslator<?> translator = TRANSLATORS.get(transform.getClass());
- return translator;
+ return TRANSLATORS.get(transform.getClass());
}
// --------------------------------------------------------------------------------------------
@@ -123,7 +122,7 @@ public class FlinkStreamingTransformTranslators {
public void translateNode(Read.Unbounded<T> transform, FlinkStreamingTranslationContext context) {
PCollection<T> output = context.getOutput(transform);
- DataStream<WindowedValue<T>> source = null;
+ DataStream<WindowedValue<T>> source;
if (transform.getSource().getClass().equals(UnboundedFlinkSource.class)) {
UnboundedFlinkSource flinkSource = (UnboundedFlinkSource) transform.getSource();
source = context.getExecutionEnvironment()
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingTranslationContext.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingTranslationContext.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingTranslationContext.java
index 83ea575..df68e50 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingTranslationContext.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/FlinkStreamingTranslationContext.java
@@ -74,13 +74,11 @@ public class FlinkStreamingTranslationContext {
@SuppressWarnings("unchecked")
public <I extends PInput> I getInput(PTransform<I, ?> transform) {
- I input = (I) currentTransform.getInput();
- return input;
+ return (I) currentTransform.getInput();
}
@SuppressWarnings("unchecked")
public <O extends POutput> O getOutput(PTransform<?, O> transform) {
- O output = (O) currentTransform.getOutput();
- return output;
+ return (O) currentTransform.getOutput();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/functions/FlinkCoGroupKeyedListAggregator.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/functions/FlinkCoGroupKeyedListAggregator.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/functions/FlinkCoGroupKeyedListAggregator.java
index f859348..4c7fefd 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/functions/FlinkCoGroupKeyedListAggregator.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/functions/FlinkCoGroupKeyedListAggregator.java
@@ -53,6 +53,6 @@ public class FlinkCoGroupKeyedListAggregator<K,V1,V2> implements CoGroupFunction
k = entry.getKey();
result.add(new RawUnionValue(index2, entry.getValue()));
}
- out.collect(KV.of(k, new CoGbkResult(schema, (List) result)));
+ out.collect(KV.of(k, new CoGbkResult(schema, result)));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/wrappers/SinkOutputFormat.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/wrappers/SinkOutputFormat.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/wrappers/SinkOutputFormat.java
index b10c86f..ec8c186 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/wrappers/SinkOutputFormat.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/wrappers/SinkOutputFormat.java
@@ -23,7 +23,6 @@ import com.google.cloud.dataflow.sdk.io.Sink;
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
import com.google.cloud.dataflow.sdk.repackaged.com.google.common.base.Preconditions;
import com.google.cloud.dataflow.sdk.transforms.Write;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.io.OutputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.AbstractID;
@@ -32,7 +31,6 @@ import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Field;
-import java.util.UUID;
/**
* Wrapper class to use generic Write.Bound transforms as sinks.
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/wrappers/SourceInputFormat.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/wrappers/SourceInputFormat.java b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/wrappers/SourceInputFormat.java
index afb15da..b3eca96 100644
--- a/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/wrappers/SourceInputFormat.java
+++ b/runners/flink/src/main/java/com/dataartisans/flink/dataflow/translation/wrappers/SourceInputFormat.java
@@ -45,7 +45,6 @@ public class SourceInputFormat<T> implements InputFormat<T, SourceInputSplit<T>>
private final BoundedSource<T> initialSource;
private transient PipelineOptions options;
- private final Coder<T> coder;
private BoundedSource.BoundedReader<T> reader = null;
private boolean reachedEnd = true;
@@ -53,7 +52,7 @@ public class SourceInputFormat<T> implements InputFormat<T, SourceInputSplit<T>>
public SourceInputFormat(BoundedSource<T> initialSource, PipelineOptions options, Coder<T> coder) {
this.initialSource = initialSource;
this.options = options;
- this.coder = coder;
+ Coder<T> coder1 = coder;
}
private void writeObject(ObjectOutputStream out)
@@ -111,12 +110,12 @@ public class SourceInputFormat<T> implements InputFormat<T, SourceInputSplit<T>>
@Override
@SuppressWarnings("unchecked")
public SourceInputSplit<T>[] createInputSplits(int numSplits) throws IOException {
- long desiredSizeBytes = 10000;
+ long desiredSizeBytes;
try {
desiredSizeBytes = initialSource.getEstimatedSizeBytes(options) / numSplits;
List<? extends Source<T>> shards = initialSource.splitIntoBundles(desiredSizeBytes,
options);
- List<SourceInputSplit<T>> splits = new ArrayList<SourceInputSplit<T>>();
+ List<SourceInputSplit<T>> splits = new ArrayList<>();
int splitCount = 0;
for (Source<T> shard: shards) {
splits.add(new SourceInputSplit<>(shard, splitCount++));
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/test/java/com/dataartisans/flink/dataflow/WriteSinkITCase.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/test/java/com/dataartisans/flink/dataflow/WriteSinkITCase.java b/runners/flink/src/test/java/com/dataartisans/flink/dataflow/WriteSinkITCase.java
index c8302e8..205fe9b 100644
--- a/runners/flink/src/test/java/com/dataartisans/flink/dataflow/WriteSinkITCase.java
+++ b/runners/flink/src/test/java/com/dataartisans/flink/dataflow/WriteSinkITCase.java
@@ -32,7 +32,7 @@ import java.io.File;
import java.io.PrintWriter;
import java.net.URI;
-import static junit.framework.Assert.*;
+import static org.junit.Assert.*;
/**
* Tests the translation of custom Write.Bound sinks.
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/602d8fe9/runners/flink/src/test/java/com/dataartisans/flink/dataflow/util/JoinExamples.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/test/java/com/dataartisans/flink/dataflow/util/JoinExamples.java b/runners/flink/src/test/java/com/dataartisans/flink/dataflow/util/JoinExamples.java
index 74f754b..aa5623d 100644
--- a/runners/flink/src/test/java/com/dataartisans/flink/dataflow/util/JoinExamples.java
+++ b/runners/flink/src/test/java/com/dataartisans/flink/dataflow/util/JoinExamples.java
@@ -76,7 +76,7 @@ public class JoinExamples {
KV<String, CoGbkResult> e = c.element();
CoGbkResult val = e.getValue();
String countryCode = e.getKey();
- String countryName = "none";
+ String countryName;
countryName = e.getValue().getOnly(countryInfoTag, "Kostas");
for (String eventInfo : c.element().getValue().getAll(eventInfoTag)) {
// Generate a string that combines information from both collection values