This is an automated email from the ASF dual-hosted git repository.
nkruber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-training.git
The following commit(s) were added to refs/heads/master by this push:
new b7bc718 [FLINK-17277] address IntelliJ warnings
b7bc718 is described below
commit b7bc718bc6d178d474ebdc22e2dc2a27910d693e
Author: Nico Kruber <n...@ververica.com>
AuthorDate: Sat Apr 18 22:41:51 2020 +0200
[FLINK-17277] address IntelliJ warnings
---
.../examples/ridecount/RideCountExample.java | 4 +--
.../exercises/common/datatypes/TaxiFare.java | 23 ++++++-------
.../exercises/common/datatypes/TaxiRide.java | 39 +++++++++++-----------
.../exercises/common/sources/TaxiFareSource.java | 23 ++++++-------
.../exercises/common/sources/TaxiRideSource.java | 23 ++++++-------
.../training/exercises/common/utils/GeoUtils.java | 3 +-
.../exercises/testing/TaxiRideTestBase.java | 25 +++++++++++---
.../training/exercises/testing/TestSource.java | 24 +++++--------
config/checkstyle/checkstyle.xml | 4 +--
config/checkstyle/suppressions.xml | 4 +--
hourly-tips/DISCUSSION.md | 2 +-
.../solutions/hourlytips/HourlyTipsSolution.java | 6 ++--
.../hourlytips/scala/HourlyTipsSolution.scala | 2 +-
.../exercises/hourlytips/HourlyTipsTest.java | 14 ++++----
.../hourlytips/scala/HourlyTipsTest.scala | 2 +-
.../exercises/longrides/LongRidesTest.java | 28 ++++++++--------
.../ridecleansing/RideCleansingSolution.java | 3 +-
.../exercises/ridecleansing/RideCleansingTest.java | 4 +--
.../ridesandfares/RidesAndFaresSolution.java | 4 +--
.../exercises/ridesandfares/RidesAndFaresTest.java | 12 +++----
.../ridesandfares/scala/RidesAndFaresTest.scala | 2 +-
21 files changed, 124 insertions(+), 127 deletions(-)
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 @@ public class RideCountExample {
// 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 @@ import java.util.Locale;
*/
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 @@ public class TaxiFare implements Serializable {
@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 @@ public class TaxiFare implements Serializable {
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.DateTime;
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 @@ import java.util.Locale;
*/
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 @@ public class TaxiRide implements Comparable<TaxiRide>,
Serializable {
@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 @@ public class TaxiRide implements Comparable<TaxiRide>,
Serializable {
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 @@ public class TaxiRide implements Comparable<TaxiRide>,
Serializable {
* <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.FileInputStream;
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 @@ public class TaxiFareSource implements
SourceFunction<TaxiFare> {
}
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 class TaxiFareSource implements
SourceFunction<TaxiFare> {
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 @@ public class TaxiFareSource implements
SourceFunction<TaxiFare> {
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 @@ public class TaxiFareSource implements
SourceFunction<TaxiFare> {
// 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 @@ public class TaxiFareSource implements
SourceFunction<TaxiFare> {
) {
// 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 @@ public class TaxiFareSource implements
SourceFunction<TaxiFare> {
}
// 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 @@ public class TaxiFareSource implements
SourceFunction<TaxiFare> {
// 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.FileInputStream;
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 @@ public class TaxiRideSource implements
SourceFunction<TaxiRide> {
}
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 class TaxiRideSource implements
SourceFunction<TaxiRide> {
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 @@ public class TaxiRideSource implements
SourceFunction<TaxiRide> {
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 @@ public class TaxiRideSource implements
SourceFunction<TaxiRide> {
// 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 @@ public class TaxiRideSource implements
SourceFunction<TaxiRide> {
) {
// 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 @@ public class TaxiRideSource implements
SourceFunction<TaxiRide> {
}
// 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 @@ public class TaxiRideSource implements
SourceFunction<TaxiRide> {
// 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 @@ public class GeoUtils {
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.ArrayList;
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 abstract class TaxiRideTestBase<OUT> {
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 @@ package org.apache.flink.training.exercises.testing;
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 @@ specific language governing permissions and limitations
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 @@ under the License.
-->
<!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 @@ public static class AddTips extends ProcessWindowFunction<
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 class HourlyTipsSolution extends ExerciseBase {
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 @@ object HourlyTipsSolution {
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 @@ import static org.junit.Assert.assertEquals;
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 @@ public class HourlyTipsTest extends
TaxiRideTestBase<Tuple3<Long, Long, Float>>
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 @@ public class HourlyTipsTest extends
TaxiRideTestBase<Tuple3<Long, Long, Float>>
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 @@ public class HourlyTipsTest extends
TaxiRideTestBase<Tuple3<Long, Long, Float>>
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 @@ public class HourlyTipsTest extends
TaxiRideTestBase<Tuple3<Long, Long, Float>>
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 @@ class HourlyTipsTest extends hourlytips.HourlyTipsTest {
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 @@ import static org.junit.Assert.assertEquals;
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 @@ public class LongRidesTest extends TaxiRideTestBase<TaxiRide>
{
@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 @@ public class LongRidesTest extends TaxiRideTestBase<TaxiRide>
{
@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 @@ public class LongRidesTest extends TaxiRideTestBase<TaxiRide>
{
@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 @@ public class LongRidesTest extends TaxiRideTestBase<TaxiRide>
{
@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 @@ public class LongRidesTest extends
TaxiRideTestBase<TaxiRide> {
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 class RideCleansingSolution extends ExerciseBase {
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 @@ import static org.junit.Assert.assertEquals;
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 @@ public class RideCleansingTest extends
TaxiRideTestBase<TaxiRide> {
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 @@ public class RidesAndFaresSolution extends ExerciseBase {
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 @@ public class RidesAndFaresSolution extends ExerciseBase {
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 @@ import static org.junit.Assert.assertThat;
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 @@ public class RidesAndFaresTest extends
TaxiRideTestBase<Tuple2<TaxiRide, TaxiFar
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 @@ public class RidesAndFaresTest extends
TaxiRideTestBase<Tuple2<TaxiRide, TaxiFar
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 @@ public class RidesAndFaresTest extends
TaxiRideTestBase<Tuple2<TaxiRide, TaxiFar
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 @@ class RidesAndFaresTest extends
ridesandfares.RidesAndFaresTest{
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]] = {
