diff --git a/common/src/main/java/org/apache/flink/training/examples/ridecount/RideCountExample.java b/common/src/main/java/org/apache/flink/training/examples/ridecount/RideCountExample.java
index fe17f93..ac7424a 100644
--- a/common/src/main/java/org/apache/flink/training/examples/ridecount/RideCountExample.java
+++ b/common/src/main/java/org/apache/flink/training/examples/ridecount/RideCountExample.java
@@ -67,8 +67,8 @@
 		// map each ride to a tuple of (driverId, 1)
 		DataStream<Tuple2<Long, Long>> tuples = rides.map(new MapFunction<TaxiRide, Tuple2<Long, Long>>() {
 					@Override
-					public Tuple2<Long, Long> map(TaxiRide ride) throws Exception {
-						return new Tuple2<Long, Long>(ride.driverId, 1L);
+					public Tuple2<Long, Long> map(TaxiRide ride) {
+						return Tuple2.of(ride.driverId, 1L);
 					}
 		});
 
diff --git a/common/src/main/java/org/apache/flink/training/exercises/common/datatypes/TaxiFare.java b/common/src/main/java/org/apache/flink/training/exercises/common/datatypes/TaxiFare.java
index 393638e..f1da3e6 100644
--- a/common/src/main/java/org/apache/flink/training/exercises/common/datatypes/TaxiFare.java
+++ b/common/src/main/java/org/apache/flink/training/exercises/common/datatypes/TaxiFare.java
@@ -34,7 +34,7 @@
  */
 public class TaxiFare implements Serializable {
 
-	private static transient DateTimeFormatter timeFormatter =
+	private static final DateTimeFormatter TIME_FORMATTER =
 			DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss").withLocale(Locale.US).withZoneUTC();
 
 	/**
@@ -69,17 +69,14 @@
 
 	@Override
 	public String toString() {
-		StringBuilder sb = new StringBuilder();
-		sb.append(rideId).append(",");
-		sb.append(taxiId).append(",");
-		sb.append(driverId).append(",");
-		sb.append(startTime.toString(timeFormatter)).append(",");
-		sb.append(paymentType).append(",");
-		sb.append(tip).append(",");
-		sb.append(tolls).append(",");
-		sb.append(totalFare);
-
-		return sb.toString();
+		return rideId + "," +
+				taxiId + "," +
+				driverId + "," +
+				startTime.toString(TIME_FORMATTER) + "," +
+				paymentType + "," +
+				tip + "," +
+				tolls + "," +
+				totalFare;
 	}
 
 	/**
@@ -98,7 +95,7 @@
 			ride.rideId = Long.parseLong(tokens[0]);
 			ride.taxiId = Long.parseLong(tokens[1]);
 			ride.driverId = Long.parseLong(tokens[2]);
-			ride.startTime = DateTime.parse(tokens[3], timeFormatter);
+			ride.startTime = DateTime.parse(tokens[3], TIME_FORMATTER);
 			ride.paymentType = tokens[4];
 			ride.tip = tokens[5].length() > 0 ? Float.parseFloat(tokens[5]) : 0.0f;
 			ride.tolls = tokens[6].length() > 0 ? Float.parseFloat(tokens[6]) : 0.0f;
diff --git a/common/src/main/java/org/apache/flink/training/exercises/common/datatypes/TaxiRide.java b/common/src/main/java/org/apache/flink/training/exercises/common/datatypes/TaxiRide.java
index 7c78dec..a923a00 100644
--- a/common/src/main/java/org/apache/flink/training/exercises/common/datatypes/TaxiRide.java
+++ b/common/src/main/java/org/apache/flink/training/exercises/common/datatypes/TaxiRide.java
@@ -24,6 +24,8 @@
 import org.joda.time.format.DateTimeFormat;
 import org.joda.time.format.DateTimeFormatter;
 
+import javax.annotation.Nullable;
+
 import java.io.Serializable;
 import java.util.Locale;
 
@@ -45,7 +47,7 @@
  */
 public class TaxiRide implements Comparable<TaxiRide>, Serializable {
 
-	private static transient DateTimeFormatter timeFormatter =
+	private static final DateTimeFormatter TIME_FORMATTER =
 			DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss").withLocale(Locale.US).withZoneUTC();
 
 	/**
@@ -89,20 +91,17 @@
 
 	@Override
 	public String toString() {
-		StringBuilder sb = new StringBuilder();
-		sb.append(rideId).append(",");
-		sb.append(isStart ? "START" : "END").append(",");
-		sb.append(startTime.toString(timeFormatter)).append(",");
-		sb.append(endTime.toString(timeFormatter)).append(",");
-		sb.append(startLon).append(",");
-		sb.append(startLat).append(",");
-		sb.append(endLon).append(",");
-		sb.append(endLat).append(",");
-		sb.append(passengerCnt).append(",");
-		sb.append(taxiId).append(",");
-		sb.append(driverId);
-
-		return sb.toString();
+		return rideId + "," +
+				(isStart ? "START" : "END") + "," +
+				startTime.toString(TIME_FORMATTER) + "," +
+				endTime.toString(TIME_FORMATTER) + "," +
+				startLon + "," +
+				startLat + "," +
+				endLon + "," +
+				endLat + "," +
+				passengerCnt + "," +
+				taxiId + "," +
+				driverId;
 	}
 
 	/**
@@ -123,13 +122,13 @@
 			switch (tokens[1]) {
 				case "START":
 					ride.isStart = true;
-					ride.startTime = DateTime.parse(tokens[2], timeFormatter);
-					ride.endTime = DateTime.parse(tokens[3], timeFormatter);
+					ride.startTime = DateTime.parse(tokens[2], TIME_FORMATTER);
+					ride.endTime = DateTime.parse(tokens[3], TIME_FORMATTER);
 					break;
 				case "END":
 					ride.isStart = false;
-					ride.endTime = DateTime.parse(tokens[2], timeFormatter);
-					ride.startTime = DateTime.parse(tokens[3], timeFormatter);
+					ride.endTime = DateTime.parse(tokens[2], TIME_FORMATTER);
+					ride.startTime = DateTime.parse(tokens[3], TIME_FORMATTER);
 					break;
 				default:
 					throw new RuntimeException("Invalid record: " + line);
@@ -158,7 +157,7 @@
 	 *     <li>putting START events before END events if they have the same timestamp</li>
 	 * </ul>
 	 */
-	public int compareTo(TaxiRide other) {
+	public int compareTo(@Nullable TaxiRide other) {
 		if (other == null) {
 			return 1;
 		}
diff --git a/common/src/main/java/org/apache/flink/training/exercises/common/sources/TaxiFareSource.java b/common/src/main/java/org/apache/flink/training/exercises/common/sources/TaxiFareSource.java
index c0f3c88..a02eff6 100644
--- a/common/src/main/java/org/apache/flink/training/exercises/common/sources/TaxiFareSource.java
+++ b/common/src/main/java/org/apache/flink/training/exercises/common/sources/TaxiFareSource.java
@@ -28,6 +28,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
 import java.util.Calendar;
 import java.util.Comparator;
 import java.util.PriorityQueue;
@@ -103,7 +104,7 @@
 		}
 		this.dataFilePath = dataFilePath;
 		this.maxDelayMsecs = maxEventDelaySecs * 1000;
-		this.watermarkDelayMSecs = maxDelayMsecs < 10000 ? 10000 : maxDelayMsecs;
+		this.watermarkDelayMSecs = Math.max(maxDelayMsecs, 10000);
 		this.servingSpeed = servingSpeedFactor;
 	}
 
@@ -111,7 +112,7 @@
 	public void run(SourceContext<TaxiFare> sourceContext) throws Exception {
 
 		gzipStream = new GZIPInputStream(new FileInputStream(dataFilePath));
-		reader = new BufferedReader(new InputStreamReader(gzipStream, "UTF-8"));
+		reader = new BufferedReader(new InputStreamReader(gzipStream, StandardCharsets.UTF_8));
 
 		generateUnorderedStream(sourceContext);
 
@@ -130,12 +131,7 @@
 		Random rand = new Random(7452);
 		PriorityQueue<Tuple2<Long, Object>> emitSchedule = new PriorityQueue<>(
 				32,
-				new Comparator<Tuple2<Long, Object>>() {
-					@Override
-					public int compare(Tuple2<Long, Object> o1, Tuple2<Long, Object> o2) {
-						return o1.f0.compareTo(o2.f0);
-					}
-				});
+				Comparator.comparing(o -> o.f0));
 
 		// read first ride and insert it into emit schedule
 		String line;
@@ -148,11 +144,11 @@
 			// get delayed time
 			long delayedEventTime = dataStartTime + getNormalDelayMsecs(rand);
 
-			emitSchedule.add(new Tuple2<Long, Object>(delayedEventTime, fare));
+			emitSchedule.add(Tuple2.of(delayedEventTime, fare));
 			// schedule next watermark
 			long watermarkTime = dataStartTime + watermarkDelayMSecs;
 			Watermark nextWatermark = new Watermark(watermarkTime - maxDelayMsecs - 1);
-			emitSchedule.add(new Tuple2<Long, Object>(watermarkTime, nextWatermark));
+			emitSchedule.add(Tuple2.of(watermarkTime, nextWatermark));
 
 		} else {
 			return;
@@ -176,7 +172,7 @@
 					) {
 				// insert event into emit schedule
 				long delayedEventTime = rideEventTime + getNormalDelayMsecs(rand);
-				emitSchedule.add(new Tuple2<Long, Object>(delayedEventTime, fare));
+				emitSchedule.add(Tuple2.of(delayedEventTime, fare));
 
 				// read next ride
 				if (reader.ready() && (line = reader.readLine()) != null) {
@@ -190,13 +186,14 @@
 			}
 
 			// emit schedule is updated, emit next element in schedule
-			Tuple2<Long, Object> head = emitSchedule.poll();
+			Tuple2<Long, Object> head = emitSchedule.remove();
 			long delayedEventTime = head.f0;
 
 			long now = Calendar.getInstance().getTimeInMillis();
 			long servingTime = toServingTime(servingStartTime, dataStartTime, delayedEventTime);
 			long waitTime = servingTime - now;
 
+			//noinspection BusyWait
 			Thread.sleep((waitTime > 0) ? waitTime : 0);
 
 			if (head.f1 instanceof TaxiFare) {
@@ -211,7 +208,7 @@
 				// schedule next watermark
 				long watermarkTime = delayedEventTime + watermarkDelayMSecs;
 				Watermark nextWatermark = new Watermark(watermarkTime - maxDelayMsecs - 1);
-				emitSchedule.add(new Tuple2<Long, Object>(watermarkTime, nextWatermark));
+				emitSchedule.add(Tuple2.of(watermarkTime, nextWatermark));
 			}
 		}
 	}
diff --git a/common/src/main/java/org/apache/flink/training/exercises/common/sources/TaxiRideSource.java b/common/src/main/java/org/apache/flink/training/exercises/common/sources/TaxiRideSource.java
index 6bfb970..b19a310 100644
--- a/common/src/main/java/org/apache/flink/training/exercises/common/sources/TaxiRideSource.java
+++ b/common/src/main/java/org/apache/flink/training/exercises/common/sources/TaxiRideSource.java
@@ -28,6 +28,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
 import java.util.Calendar;
 import java.util.Comparator;
 import java.util.PriorityQueue;
@@ -103,7 +104,7 @@
 		}
 		this.dataFilePath = dataFilePath;
 		this.maxDelayMsecs = maxEventDelaySecs * 1000;
-		this.watermarkDelayMSecs = maxDelayMsecs < 10000 ? 10000 : maxDelayMsecs;
+		this.watermarkDelayMSecs = Math.max(maxDelayMsecs, 10000);
 		this.servingSpeed = servingSpeedFactor;
 	}
 
@@ -111,7 +112,7 @@
 	public void run(SourceContext<TaxiRide> sourceContext) throws Exception {
 
 		gzipStream = new GZIPInputStream(new FileInputStream(dataFilePath));
-		reader = new BufferedReader(new InputStreamReader(gzipStream, "UTF-8"));
+		reader = new BufferedReader(new InputStreamReader(gzipStream, StandardCharsets.UTF_8));
 
 		generateUnorderedStream(sourceContext);
 
@@ -130,12 +131,7 @@
 		Random rand = new Random(7452);
 		PriorityQueue<Tuple2<Long, Object>> emitSchedule = new PriorityQueue<>(
 				32,
-				new Comparator<Tuple2<Long, Object>>() {
-					@Override
-					public int compare(Tuple2<Long, Object> o1, Tuple2<Long, Object> o2) {
-						return o1.f0.compareTo(o2.f0);
-					}
-				});
+				Comparator.comparing(o -> o.f0));
 
 		// read first ride and insert it into emit schedule
 		String line;
@@ -148,11 +144,11 @@
 			// get delayed time
 			long delayedEventTime = dataStartTime + getNormalDelayMsecs(rand);
 
-			emitSchedule.add(new Tuple2<Long, Object>(delayedEventTime, ride));
+			emitSchedule.add(Tuple2.of(delayedEventTime, ride));
 			// schedule next watermark
 			long watermarkTime = dataStartTime + watermarkDelayMSecs;
 			Watermark nextWatermark = new Watermark(watermarkTime - maxDelayMsecs - 1);
-			emitSchedule.add(new Tuple2<Long, Object>(watermarkTime, nextWatermark));
+			emitSchedule.add(Tuple2.of(watermarkTime, nextWatermark));
 
 		} else {
 			return;
@@ -176,7 +172,7 @@
 					) {
 				// insert event into emit schedule
 				long delayedEventTime = rideEventTime + getNormalDelayMsecs(rand);
-				emitSchedule.add(new Tuple2<Long, Object>(delayedEventTime, ride));
+				emitSchedule.add(Tuple2.of(delayedEventTime, ride));
 
 				// read next ride
 				if (reader.ready() && (line = reader.readLine()) != null) {
@@ -190,13 +186,14 @@
 			}
 
 			// emit schedule is updated, emit next element in schedule
-			Tuple2<Long, Object> head = emitSchedule.poll();
+			Tuple2<Long, Object> head = emitSchedule.remove();
 			long delayedEventTime = head.f0;
 
 			long now = Calendar.getInstance().getTimeInMillis();
 			long servingTime = toServingTime(servingStartTime, dataStartTime, delayedEventTime);
 			long waitTime = servingTime - now;
 
+			//noinspection BusyWait
 			Thread.sleep((waitTime > 0) ? waitTime : 0);
 
 			if (head.f1 instanceof TaxiRide) {
@@ -211,7 +208,7 @@
 				// schedule next watermark
 				long watermarkTime = delayedEventTime + watermarkDelayMSecs;
 				Watermark nextWatermark = new Watermark(watermarkTime - maxDelayMsecs - 1);
-				emitSchedule.add(new Tuple2<Long, Object>(watermarkTime, nextWatermark));
+				emitSchedule.add(Tuple2.of(watermarkTime, nextWatermark));
 			}
 		}
 	}
diff --git a/common/src/main/java/org/apache/flink/training/exercises/common/utils/GeoUtils.java b/common/src/main/java/org/apache/flink/training/exercises/common/utils/GeoUtils.java
index 4594248..e638ab3 100644
--- a/common/src/main/java/org/apache/flink/training/exercises/common/utils/GeoUtils.java
+++ b/common/src/main/java/org/apache/flink/training/exercises/common/utils/GeoUtils.java
@@ -245,9 +245,8 @@
 
 		double x = destLat - startLat;
 		double y = (destLon - startLon) * Math.cos(startLat);
-		int degrees = (int) Math.toDegrees(Math.atan2(x, y)) + 179;
 
-		return degrees;
+		return (int) Math.toDegrees(Math.atan2(x, y)) + 179;
 	}
 
 }
diff --git a/common/src/test/java/org/apache/flink/training/exercises/testing/TaxiRideTestBase.java b/common/src/test/java/org/apache/flink/training/exercises/testing/TaxiRideTestBase.java
index 6c59de1..f4f5596 100644
--- a/common/src/test/java/org/apache/flink/training/exercises/testing/TaxiRideTestBase.java
+++ b/common/src/test/java/org/apache/flink/training/exercises/testing/TaxiRideTestBase.java
@@ -30,34 +30,49 @@
 import java.util.List;
 
 public abstract class TaxiRideTestBase<OUT> {
-	public static class TestRideSource extends TestSource implements ResultTypeQueryable<TaxiRide> {
+	public static class TestRideSource extends TestSource<TaxiRide> implements ResultTypeQueryable<TaxiRide> {
 		public TestRideSource(Object ... eventsOrWatermarks) {
 			this.testStream = eventsOrWatermarks;
 		}
 
 		@Override
+		long getTimestamp(TaxiRide ride) {
+			return ride.getEventTime();
+		}
+
+		@Override
 		public TypeInformation<TaxiRide> getProducedType() {
 			return TypeInformation.of(TaxiRide.class);
 		}
 	}
 
-	public static class TestFareSource extends TestSource implements ResultTypeQueryable<TaxiFare> {
+	public static class TestFareSource extends TestSource<TaxiFare> implements ResultTypeQueryable<TaxiFare> {
 		public TestFareSource(Object ... eventsOrWatermarks) {
 			this.testStream = eventsOrWatermarks;
 		}
 
 		@Override
+		long getTimestamp(TaxiFare fare) {
+			return fare.getEventTime();
+		}
+
+		@Override
 		public TypeInformation<TaxiFare> getProducedType() {
 			return TypeInformation.of(TaxiFare.class);
 		}
 	}
 
-	public static class TestStringSource extends TestSource implements ResultTypeQueryable<String> {
+	public static class TestStringSource extends TestSource<String> implements ResultTypeQueryable<String> {
 		public TestStringSource(Object ... eventsOrWatermarks) {
 			this.testStream = eventsOrWatermarks;
 		}
 
 		@Override
+		long getTimestamp(String s) {
+			return 0L;
+		}
+
+		@Override
 		public TypeInformation<String> getProducedType() {
 			return TypeInformation.of(String.class);
 		}
@@ -69,13 +84,13 @@
 		public static final List VALUES = new ArrayList<>();
 
 		@Override
-		public void invoke(OUT value, Context context) throws Exception {
+		public void invoke(OUT value, Context context) {
 			VALUES.add(value);
 		}
 	}
 
 	public interface Testable {
-		public abstract void main() throws Exception;
+		void main() throws Exception;
 	}
 
 	protected List<OUT> runApp(TestRideSource source, TestSink<OUT> sink, Testable exercise, Testable solution) throws Exception {
diff --git a/common/src/test/java/org/apache/flink/training/exercises/testing/TestSource.java b/common/src/test/java/org/apache/flink/training/exercises/testing/TestSource.java
index 57d72ab..84ba53e 100644
--- a/common/src/test/java/org/apache/flink/training/exercises/testing/TestSource.java
+++ b/common/src/test/java/org/apache/flink/training/exercises/testing/TestSource.java
@@ -20,35 +20,29 @@
 
 import org.apache.flink.streaming.api.functions.source.SourceFunction;
 import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.training.exercises.common.datatypes.TaxiFare;
-import org.apache.flink.training.exercises.common.datatypes.TaxiRide;
 
-public abstract class TestSource implements SourceFunction {
+public abstract class TestSource<T> implements SourceFunction<T> {
 	private volatile boolean running = true;
+	// T or watermark (Long)
 	protected Object[] testStream;
 
 	@Override
-	public void run(SourceContext ctx) throws Exception {
+	public void run(SourceContext<T> ctx) {
 		for (int i = 0; (i < testStream.length) && running; i++) {
-			if (testStream[i] instanceof TaxiRide) {
-				TaxiRide ride = (TaxiRide) testStream[i];
-				ctx.collectWithTimestamp(ride, ride.getEventTime());
-			} else if (testStream[i] instanceof TaxiFare) {
-				TaxiFare fare = (TaxiFare) testStream[i];
-				ctx.collectWithTimestamp(fare, fare.getEventTime());
-			} else if (testStream[i] instanceof String) {
-				String s = (String) testStream[i];
-				ctx.collectWithTimestamp(s, 0);
-			} else if (testStream[i] instanceof Long) {
+			if (testStream[i] instanceof Long) {
 				Long ts = (Long) testStream[i];
 				ctx.emitWatermark(new Watermark(ts));
 			} else {
-				throw new RuntimeException(testStream[i].toString());
+				//noinspection unchecked
+				T element = (T) testStream[i];
+				ctx.collectWithTimestamp(element, getTimestamp(element));
 			}
 		}
 		// test sources are finite, so they have a Long.MAX_VALUE watermark when they finishes
 	}
 
+	abstract long getTimestamp(T element);
+
 	@Override
 	public void cancel() {
 		running = false;
diff --git a/config/checkstyle/checkstyle.xml b/config/checkstyle/checkstyle.xml
index 34b0e76..e656fda 100644
--- a/config/checkstyle/checkstyle.xml
+++ b/config/checkstyle/checkstyle.xml
@@ -18,8 +18,8 @@
 under the License.
 -->
 <!DOCTYPE module PUBLIC
-          "-//Puppy Crawl//DTD Check Configuration 1.3//EN"
-          "http://www.puppycrawl.com/dtds/configuration_1_3.dtd">
+        "-//Checkstyle//DTD Checkstyle Configuration 1.3//EN"
+        "https://checkstyle.org/dtds/configuration_1_3.dtd">
 
 <!--
 This is a checkstyle configuration file. For descriptions of
diff --git a/config/checkstyle/suppressions.xml b/config/checkstyle/suppressions.xml
index 0b05a64..01e36b5 100644
--- a/config/checkstyle/suppressions.xml
+++ b/config/checkstyle/suppressions.xml
@@ -19,8 +19,8 @@
 -->
 
 <!DOCTYPE suppressions PUBLIC
-		"-//Puppy Crawl//DTD Suppressions 1.1//EN"
-		"http://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
+		"-//Checkstyle//DTD SuppressionFilter Configuration 1.2//EN"
+		"https://checkstyle.org/dtds/suppressions_1_2.dtd">
 
 <suppressions>
 	<suppress
diff --git a/hourly-tips/DISCUSSION.md b/hourly-tips/DISCUSSION.md
index 2e91b2d..ab653d3 100644
--- a/hourly-tips/DISCUSSION.md
+++ b/hourly-tips/DISCUSSION.md
@@ -41,7 +41,7 @@
 		for (TaxiFare f : fares) {
 			sumOfTips += f.tip;
 		}
-		out.collect(new Tuple3<>(context.window().getEnd(), key, sumOfTips));
+		out.collect(Tuple3.of(context.window().getEnd(), key, sumOfTips));
 	}
 }
 ```
diff --git a/hourly-tips/src/solution/java/org/apache/flink/training/solutions/hourlytips/HourlyTipsSolution.java b/hourly-tips/src/solution/java/org/apache/flink/training/solutions/hourlytips/HourlyTipsSolution.java
index 4e2bb32..b640a19 100644
--- a/hourly-tips/src/solution/java/org/apache/flink/training/solutions/hourlytips/HourlyTipsSolution.java
+++ b/hourly-tips/src/solution/java/org/apache/flink/training/solutions/hourlytips/HourlyTipsSolution.java
@@ -96,12 +96,12 @@
 	public static class AddTips extends ProcessWindowFunction<
 			TaxiFare, Tuple3<Long, Long, Float>, Long, TimeWindow> {
 		@Override
-		public void process(Long key, Context context, Iterable<TaxiFare> fares, Collector<Tuple3<Long, Long, Float>> out) throws Exception {
-			Float sumOfTips = 0F;
+		public void process(Long key, Context context, Iterable<TaxiFare> fares, Collector<Tuple3<Long, Long, Float>> out) {
+			float sumOfTips = 0F;
 			for (TaxiFare f : fares) {
 				sumOfTips += f.tip;
 			}
-			out.collect(new Tuple3<>(context.window().getEnd(), key, sumOfTips));
+			out.collect(Tuple3.of(context.window().getEnd(), key, sumOfTips));
 		}
 	}
 }
diff --git a/hourly-tips/src/solution/scala/org/apache/flink/training/solutions/hourlytips/scala/HourlyTipsSolution.scala b/hourly-tips/src/solution/scala/org/apache/flink/training/solutions/hourlytips/scala/HourlyTipsSolution.scala
index 3ad6243..ab3de6c 100644
--- a/hourly-tips/src/solution/scala/org/apache/flink/training/solutions/hourlytips/scala/HourlyTipsSolution.scala
+++ b/hourly-tips/src/solution/scala/org/apache/flink/training/solutions/hourlytips/scala/HourlyTipsSolution.scala
@@ -82,7 +82,7 @@
   class WrapWithWindowInfo() extends ProcessWindowFunction[(Long, Float), (Long, Long, Float), Long, TimeWindow] {
     override def process(key: Long, context: Context, elements: Iterable[(Long, Float)], out: Collector[(Long, Long, Float)]): Unit = {
       val sumOfTips = elements.iterator.next()._2
-      out.collect((context.window.getEnd(), key, sumOfTips))
+      out.collect((context.window.getEnd, key, sumOfTips))
     }
   }
 
diff --git a/hourly-tips/src/test/java/org/apache/flink/training/exercises/hourlytips/HourlyTipsTest.java b/hourly-tips/src/test/java/org/apache/flink/training/exercises/hourlytips/HourlyTipsTest.java
index 565a8d3..eef1bdf 100644
--- a/hourly-tips/src/test/java/org/apache/flink/training/exercises/hourlytips/HourlyTipsTest.java
+++ b/hourly-tips/src/test/java/org/apache/flink/training/exercises/hourlytips/HourlyTipsTest.java
@@ -35,7 +35,7 @@
 
 public class HourlyTipsTest extends TaxiRideTestBase<Tuple3<Long, Long, Float>> {
 
-	static Testable javaExercise = () -> HourlyTipsExercise.main(new String[]{});
+	static final Testable JAVA_EXERCISE = () -> HourlyTipsExercise.main(new String[]{});
 
 	@Test
 	public void testOneDriverOneTip() throws Exception {
@@ -45,7 +45,7 @@
 				one
 		);
 
-		Tuple3<Long, Long, Float> max = new Tuple3<Long, Long, Float>(t(60), 1L, 1.0F);
+		Tuple3<Long, Long, Float> max = Tuple3.of(t(60), 1L, 1.0F);
 
 		List<Tuple3<Long, Long, Float>> expected = Collections.singletonList(max);
 
@@ -64,8 +64,8 @@
 				tenIn2
 		);
 
-		Tuple3<Long, Long, Float> hour1 = new Tuple3<Long, Long, Float>(t(60), 1L, 6.0F);
-		Tuple3<Long, Long, Float> hour2 = new Tuple3<Long, Long, Float>(t(120), 1L, 10.0F);
+		Tuple3<Long, Long, Float> hour1 = Tuple3.of(t(60), 1L, 6.0F);
+		Tuple3<Long, Long, Float> hour2 = Tuple3.of(t(120), 1L, 10.0F);
 
 		List<Tuple3<Long, Long, Float>> expected = Arrays.asList(hour1, hour2);
 
@@ -86,8 +86,8 @@
 				twentyFor2In2
 		);
 
-		Tuple3<Long, Long, Float> hour1 = new Tuple3<Long, Long, Float>(t(60), 1L, 6.0F);
-		Tuple3<Long, Long, Float> hour2 = new Tuple3<Long, Long, Float>(t(120), 2L, 20.0F);
+		Tuple3<Long, Long, Float> hour1 = Tuple3.of(t(60), 1L, 6.0F);
+		Tuple3<Long, Long, Float> hour2 = Tuple3.of(t(120), 2L, 20.0F);
 
 		List<Tuple3<Long, Long, Float>> expected = Arrays.asList(hour1, hour2);
 
@@ -104,7 +104,7 @@
 
 	protected List<Tuple3<Long, Long, Float>> results(TestFareSource source) throws Exception {
 		Testable javaSolution = () -> HourlyTipsSolution.main(new String[]{});
-		return runApp(source, new TestSink<>(), javaExercise, javaSolution);
+		return runApp(source, new TestSink<>(), JAVA_EXERCISE, javaSolution);
 	}
 
 }
diff --git a/hourly-tips/src/test/scala/org/apache/flink/training/exercises/hourlytips/scala/HourlyTipsTest.scala b/hourly-tips/src/test/scala/org/apache/flink/training/exercises/hourlytips/scala/HourlyTipsTest.scala
index fbcedf8..aacaab5 100644
--- a/hourly-tips/src/test/scala/org/apache/flink/training/exercises/hourlytips/scala/HourlyTipsTest.scala
+++ b/hourly-tips/src/test/scala/org/apache/flink/training/exercises/hourlytips/scala/HourlyTipsTest.scala
@@ -32,7 +32,7 @@
   override protected def results(source: TaxiRideTestBase.TestFareSource): util.List[tuple.Tuple3[java.lang.Long, java.lang.Long, java.lang.Float]] = {
     val scalaSolution: TaxiRideTestBase.Testable = () => HourlyTipsSolution.main(Array.empty[String])
     val tuples: util.List[_] = runApp(source, new TaxiRideTestBase.TestSink[tuple.Tuple3[java.lang.Long, java.lang.Long, java.lang.Float]], scalaExercise, scalaSolution)
-    javaTuples(tuples.asInstanceOf[util.List[Tuple3[Long, Long, Float]]])
+    javaTuples(tuples.asInstanceOf[util.List[(Long, Long, Float)]])
   }
 
   private def javaTuples(a: util.List[(Long, Long, Float)]): util.ArrayList[tuple.Tuple3[java.lang.Long, java.lang.Long, java.lang.Float]] = {
diff --git a/long-ride-alerts/src/test/java/org/apache/flink/training/exercises/longrides/LongRidesTest.java b/long-ride-alerts/src/test/java/org/apache/flink/training/exercises/longrides/LongRidesTest.java
index 9bd68ae..ba3e6b6 100644
--- a/long-ride-alerts/src/test/java/org/apache/flink/training/exercises/longrides/LongRidesTest.java
+++ b/long-ride-alerts/src/test/java/org/apache/flink/training/exercises/longrides/LongRidesTest.java
@@ -32,14 +32,14 @@
 
 public class LongRidesTest extends TaxiRideTestBase<TaxiRide> {
 
-	static Testable javaExercise = () -> LongRidesExercise.main(new String[]{});
+	static final Testable JAVA_EXERCISE = () -> LongRidesExercise.main(new String[]{});
 
-	private DateTime beginning = new DateTime(2000, 1, 1, 0, 0);
+	private static final DateTime BEGINNING = new DateTime(2000, 1, 1, 0, 0);
 
 	@Test
 	public void shortRide() throws Exception {
-		DateTime oneMinLater = beginning.plusMinutes(1);
-		TaxiRide rideStarted = startRide(1, beginning);
+		DateTime oneMinLater = BEGINNING.plusMinutes(1);
+		TaxiRide rideStarted = startRide(1, BEGINNING);
 		TaxiRide endedOneMinLater = endRide(rideStarted, oneMinLater);
 		Long markOneMinLater = oneMinLater.getMillis();
 
@@ -49,8 +49,8 @@
 
 	@Test
 	public void outOfOrder() throws Exception {
-		DateTime oneMinLater = beginning.plusMinutes(1);
-		TaxiRide rideStarted = startRide(1, beginning);
+		DateTime oneMinLater = BEGINNING.plusMinutes(1);
+		TaxiRide rideStarted = startRide(1, BEGINNING);
 		TaxiRide endedOneMinLater = endRide(rideStarted, oneMinLater);
 		Long markOneMinLater = oneMinLater.getMillis();
 
@@ -60,8 +60,8 @@
 
 	@Test
 	public void noStartShort() throws Exception {
-		DateTime oneMinLater = beginning.plusMinutes(1);
-		TaxiRide rideStarted = startRide(1, beginning);
+		DateTime oneMinLater = BEGINNING.plusMinutes(1);
+		TaxiRide rideStarted = startRide(1, BEGINNING);
 		TaxiRide endedOneMinLater = endRide(rideStarted, oneMinLater);
 		Long markOneMinLater = oneMinLater.getMillis();
 
@@ -71,8 +71,8 @@
 
 	@Test
 	public void noEnd() throws Exception {
-		TaxiRide rideStarted = startRide(1, beginning);
-		Long markThreeHoursLater = beginning.plusHours(3).getMillis();
+		TaxiRide rideStarted = startRide(1, BEGINNING);
+		Long markThreeHoursLater = BEGINNING.plusHours(3).getMillis();
 
 		TestRideSource source = new TestRideSource(rideStarted, markThreeHoursLater);
 		assertEquals(Collections.singletonList(rideStarted), results(source));
@@ -80,9 +80,9 @@
 
 	@Test
 	public void longRide() throws Exception {
-		TaxiRide rideStarted = startRide(1, beginning);
-		Long mark2HoursLater = beginning.plusMinutes(120).getMillis();
-		TaxiRide rideEnded3HoursLater = endRide(rideStarted, beginning.plusHours(3));
+		TaxiRide rideStarted = startRide(1, BEGINNING);
+		Long mark2HoursLater = BEGINNING.plusMinutes(120).getMillis();
+		TaxiRide rideEnded3HoursLater = endRide(rideStarted, BEGINNING.plusHours(3));
 
 		TestRideSource source = new TestRideSource(rideStarted, mark2HoursLater, rideEnded3HoursLater);
 		assertEquals(Collections.singletonList(rideStarted), results(source));
@@ -102,7 +102,7 @@
 
 	protected List<TaxiRide> results(TestRideSource source) throws Exception {
 		Testable javaSolution = () -> LongRidesSolution.main(new String[]{});
-		return runApp(source, new TestSink<>(), javaExercise, javaSolution);
+		return runApp(source, new TestSink<>(), JAVA_EXERCISE, javaSolution);
 	}
 
 }
diff --git a/ride-cleansing/src/solution/java/org/apache/flink/training/solutions/ridecleansing/RideCleansingSolution.java b/ride-cleansing/src/solution/java/org/apache/flink/training/solutions/ridecleansing/RideCleansingSolution.java
index cf3dca5..e16ea83 100644
--- a/ride-cleansing/src/solution/java/org/apache/flink/training/solutions/ridecleansing/RideCleansingSolution.java
+++ b/ride-cleansing/src/solution/java/org/apache/flink/training/solutions/ridecleansing/RideCleansingSolution.java
@@ -74,8 +74,7 @@
 
 	public static class NYCFilter implements FilterFunction<TaxiRide> {
 		@Override
-		public boolean filter(TaxiRide taxiRide) throws Exception {
-
+		public boolean filter(TaxiRide taxiRide) {
 			return GeoUtils.isInNYC(taxiRide.startLon, taxiRide.startLat) &&
 					GeoUtils.isInNYC(taxiRide.endLon, taxiRide.endLat);
 		}
diff --git a/ride-cleansing/src/test/java/org/apache/flink/training/exercises/ridecleansing/RideCleansingTest.java b/ride-cleansing/src/test/java/org/apache/flink/training/exercises/ridecleansing/RideCleansingTest.java
index b0e151f..e4554d4 100644
--- a/ride-cleansing/src/test/java/org/apache/flink/training/exercises/ridecleansing/RideCleansingTest.java
+++ b/ride-cleansing/src/test/java/org/apache/flink/training/exercises/ridecleansing/RideCleansingTest.java
@@ -32,7 +32,7 @@
 
 public class RideCleansingTest extends TaxiRideTestBase<TaxiRide> {
 
-	static Testable javaExercise = () -> RideCleansingExercise.main(new String[]{});
+	static final Testable JAVA_EXERCISE = () -> RideCleansingExercise.main(new String[]{});
 
 	@Test
 	public void testInNYC() throws Exception {
@@ -61,7 +61,7 @@
 
 	protected List<?> results(TestRideSource source) throws Exception {
 		Testable javaSolution = () -> RideCleansingSolution.main(new String[]{});
-		return runApp(source, new TestSink<>(), javaExercise, javaSolution);
+		return runApp(source, new TestSink<>(), JAVA_EXERCISE, javaSolution);
 	}
 
 }
diff --git a/rides-and-fares/src/solution/java/org/apache/flink/training/solutions/ridesandfares/RidesAndFaresSolution.java b/rides-and-fares/src/solution/java/org/apache/flink/training/solutions/ridesandfares/RidesAndFaresSolution.java
index 961aa40..0254540 100644
--- a/rides-and-fares/src/solution/java/org/apache/flink/training/solutions/ridesandfares/RidesAndFaresSolution.java
+++ b/rides-and-fares/src/solution/java/org/apache/flink/training/solutions/ridesandfares/RidesAndFaresSolution.java
@@ -114,7 +114,7 @@
 			TaxiFare fare = fareState.value();
 			if (fare != null) {
 				fareState.clear();
-				out.collect(new Tuple2(ride, fare));
+				out.collect(Tuple2.of(ride, fare));
 			} else {
 				rideState.update(ride);
 			}
@@ -125,7 +125,7 @@
 			TaxiRide ride = rideState.value();
 			if (ride != null) {
 				rideState.clear();
-				out.collect(new Tuple2(ride, fare));
+				out.collect(Tuple2.of(ride, fare));
 			} else {
 				fareState.update(fare);
 			}
diff --git a/rides-and-fares/src/test/java/org/apache/flink/training/exercises/ridesandfares/RidesAndFaresTest.java b/rides-and-fares/src/test/java/org/apache/flink/training/exercises/ridesandfares/RidesAndFaresTest.java
index e1d270a..c2b1fb1 100644
--- a/rides-and-fares/src/test/java/org/apache/flink/training/exercises/ridesandfares/RidesAndFaresTest.java
+++ b/rides-and-fares/src/test/java/org/apache/flink/training/exercises/ridesandfares/RidesAndFaresTest.java
@@ -35,7 +35,7 @@
 
 public class RidesAndFaresTest extends TaxiRideTestBase<Tuple2<TaxiRide, TaxiFare>> {
 
-	static Testable javaExercise = () -> RidesAndFaresExercise.main(new String[]{});
+	static final Testable JAVA_EXERCISE = () -> RidesAndFaresExercise.main(new String[]{});
 
 
 	final TaxiRide ride1 = testRide(1);
@@ -49,8 +49,8 @@
 		TestFareSource fares = new TestFareSource(fare1, fare2);
 
 		List<Tuple2<TaxiRide, TaxiFare>> expected = Arrays.asList(
-				new Tuple2<>(ride1, fare1),
-				new Tuple2<>(ride2, fare2));
+				Tuple2.of(ride1, fare1),
+				Tuple2.of(ride2, fare2));
 
 		assertThat("Join results don't match", results(rides, fares), containsInAnyOrder(expected.toArray()));
 	}
@@ -61,8 +61,8 @@
 		TestFareSource fares = new TestFareSource(fare2, fare1);
 
 		List<Tuple2<TaxiRide, TaxiFare>> expected = Arrays.asList(
-				new Tuple2<>(ride1, fare1),
-				new Tuple2<>(ride2, fare2));
+				Tuple2.of(ride1, fare1),
+				Tuple2.of(ride2, fare2));
 
 		assertThat("Join results don't match", results(rides, fares), containsInAnyOrder(expected.toArray()));
 	}
@@ -78,7 +78,7 @@
 
 	protected List<?> results(TestRideSource rides, TestFareSource fares) throws Exception {
 		Testable javaSolution = () -> RidesAndFaresSolution.main(new String[]{});
-		return runApp(rides, fares, new TestSink<>(), javaExercise, javaSolution);
+		return runApp(rides, fares, new TestSink<>(), JAVA_EXERCISE, javaSolution);
 	}
 
 }
diff --git a/rides-and-fares/src/test/scala/org/apache/flink/training/exercises/ridesandfares/scala/RidesAndFaresTest.scala b/rides-and-fares/src/test/scala/org/apache/flink/training/exercises/ridesandfares/scala/RidesAndFaresTest.scala
index c162122..03f7e6a 100644
--- a/rides-and-fares/src/test/scala/org/apache/flink/training/exercises/ridesandfares/scala/RidesAndFaresTest.scala
+++ b/rides-and-fares/src/test/scala/org/apache/flink/training/exercises/ridesandfares/scala/RidesAndFaresTest.scala
@@ -33,7 +33,7 @@
   override protected def results(rides: TaxiRideTestBase.TestRideSource, fares: TaxiRideTestBase.TestFareSource): util.List[tuple.Tuple2[TaxiRide, TaxiFare]] = {
     val scalaSolution: TaxiRideTestBase.Testable = () => RidesAndFaresSolution.main(Array.empty[String])
     val tuples: util.List[_] = runApp(rides, fares, new TaxiRideTestBase.TestSink[tuple.Tuple2[TaxiRide, TaxiFare]], scalaExercise, scalaSolution)
-    javaTuples(tuples.asInstanceOf[util.List[Tuple2[TaxiRide, TaxiFare]]])
+    javaTuples(tuples.asInstanceOf[util.List[(TaxiRide, TaxiFare)]])
   }
 
   private def javaTuples(a: util.List[(TaxiRide, TaxiFare)]): util.ArrayList[tuple.Tuple2[TaxiRide, TaxiFare]] = {