[FLINK-17276][checkstyle] add definitions from Flink and enforce them
- minor changes to the checkstyle profile were needed to cope with the newer
version we have here
diff --git a/README.md b/README.md
index a6451c0..764fb5c 100644
--- a/README.md
+++ b/README.md
@@ -231,9 +231,9 @@
After downloading the datasets, open the `org.apache.flink.training.exercises.common.utils.ExerciseBase` class in your IDE, and edit these two lines to point to the two taxi ride data files you have downloaded:
```java
-public final static String pathToRideData =
+public final static String PATH_TO_RIDE_DATA =
"/Users/david/stuff/flink-training/trainingData/nycTaxiRides.gz";
-public final static String pathToFareData =
+public final static String PATH_TO_FARE_DATA =
"/Users/david/stuff/flink-training/trainingData/nycTaxiFares.gz";
```
diff --git a/build.gradle b/build.gradle
index 7337d18..5e54c95 100644
--- a/build.gradle
+++ b/build.gradle
@@ -26,6 +26,7 @@
apply plugin: 'application'
}
apply plugin: 'com.github.johnrengelman.shadow'
+ apply plugin: 'checkstyle'
// artifact properties
group = 'org.apache.flink'
@@ -85,7 +86,6 @@
implementation "org.slf4j:slf4j-log4j12:${slf4jVersion}"
flinkShadowJar "joda-time:joda-time:2.7"
- flinkShadowJar "com.google.guava:guava:19.0"
if (project != project(":common")) {
implementation project(path: ':common')
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 1ccc331..fe17f93 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
@@ -29,23 +29,31 @@
import org.apache.flink.training.exercises.common.sources.TaxiRideSource;
import org.apache.flink.training.exercises.common.utils.ExerciseBase;
-
/**
* Example that counts the rides for each driver.
*
- * Parameters:
+ * <p>Parameters:
* -input path-to-input-file
*
- * Note that this is implicitly keeping state for each driver.
- * This sort of simple, non-windowed aggregation on an unbounded set of keys will use an unbounded amount of state.
- * When this is an issue, look at the SQL/Table API, or ProcessFunction, or state TTL, all of which provide
- * mechanisms for expiring state for stale keys.
+ * <p>Note that this is implicitly keeping state for each driver.
+ * This sort of simple, non-windowed aggregation on an unbounded set of keys will use an unbounded amount of state.
+ * When this is an issue, look at the SQL/Table API, or ProcessFunction, or state TTL, all of which provide
+ * mechanisms for expiring state for stale keys.
*/
public class RideCountExample {
+
+ /**
+ * Main method.
+ *
+ * <p>Parameters:
+ * -input path-to-input-file
+ *
+ * @throws Exception which occurs during job execution.
+ */
public static void main(String[] args) throws Exception {
ParameterTool params = ParameterTool.fromArgs(args);
- final String input = params.get("input", ExerciseBase.pathToRideData);
+ final String input = params.get("input", ExerciseBase.PATH_TO_RIDE_DATA);
final int maxEventDelay = 60; // events are out of order by max 60 seconds
final int servingSpeedFactor = 600; // events of 10 minutes are served every second
@@ -60,7 +68,7 @@
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) ;
+ return new Tuple2<Long, Long>(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 429717c..393638e 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
@@ -28,22 +28,26 @@
/**
* A TaxiFare is a taxi fare event.
*
- * A TaxiFare consists of
+ * <p>A TaxiFare consists of
* - the rideId of the event
* - the time of the event
- *
*/
public class TaxiFare implements Serializable {
private static transient DateTimeFormatter timeFormatter =
DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss").withLocale(Locale.US).withZoneUTC();
+ /**
+ * Creates a TaxiFare with now as start time.
+ */
public TaxiFare() {
this.startTime = new DateTime();
}
+ /**
+ * Creates a TaxiFare with the given parameters.
+ */
public TaxiFare(long rideId, long taxiId, long driverId, DateTime startTime, String paymentType, float tip, float tolls, float totalFare) {
-
this.rideId = rideId;
this.taxiId = taxiId;
this.driverId = driverId;
@@ -63,6 +67,7 @@
public float tolls;
public float totalFare;
+ @Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append(rideId).append(",");
@@ -77,6 +82,9 @@
return sb.toString();
}
+ /**
+ * Parse a TaxiFare from a CSV representation.
+ */
public static TaxiFare fromString(String line) {
String[] tokens = line.split(",");
@@ -111,9 +119,12 @@
@Override
public int hashCode() {
- return (int)this.rideId;
+ return (int) this.rideId;
}
+ /**
+ * Gets the fare's start time.
+ */
public long getEventTime() {
return startTime.getMillis();
}
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 8dea262..7c78dec 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
@@ -18,7 +18,6 @@
package org.apache.flink.training.exercises.common.datatypes;
-
import org.apache.flink.training.exercises.common.utils.GeoUtils;
import org.joda.time.DateTime;
@@ -32,7 +31,7 @@
* A TaxiRide is a taxi ride event. There are two types of events, a taxi ride start event and a
* taxi ride end event. The isStart flag specifies the type of the event.
*
- * A TaxiRide consists of
+ * <p>A TaxiRide consists of
* - the rideId of the event which is identical for start and end record
* - the type of the event (start or end)
* - the time of the event
@@ -43,22 +42,26 @@
* - the passengerCnt of the ride
* - the taxiId
* - the driverId
- *
*/
public class TaxiRide implements Comparable<TaxiRide>, Serializable {
private static transient DateTimeFormatter timeFormatter =
DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss").withLocale(Locale.US).withZoneUTC();
+ /**
+ * Creates a new TaxiRide with now as start and end time.
+ */
public TaxiRide() {
this.startTime = new DateTime();
this.endTime = new DateTime();
}
+ /**
+ * Creates a TaxiRide with the given parameters.
+ */
public TaxiRide(long rideId, boolean isStart, DateTime startTime, DateTime endTime,
float startLon, float startLat, float endLon, float endLat,
short passengerCnt, long taxiId, long driverId) {
-
this.rideId = rideId;
this.isStart = isStart;
this.startTime = startTime;
@@ -84,6 +87,7 @@
public long taxiId;
public long driverId;
+ @Override
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append(rideId).append(",");
@@ -101,6 +105,9 @@
return sb.toString();
}
+ /**
+ * Parse a TaxiRide from a CSV representation.
+ */
public static TaxiRide fromString(String line) {
String[] tokens = line.split(",");
@@ -143,8 +150,14 @@
return ride;
}
- // sort by timestamp,
- // putting START events before END events if they have the same timestamp
+ /**
+ * Compares this TaxiRide with the given one.
+ *
+ * <ul>
+ * <li>sort by timestamp,</li>
+ * <li>putting START events before END events if they have the same timestamp</li>
+ * </ul>
+ */
public int compareTo(TaxiRide other) {
if (other == null) {
return 1;
@@ -176,9 +189,12 @@
@Override
public int hashCode() {
- return (int)this.rideId;
+ return (int) this.rideId;
}
+ /**
+ * Gets the ride's time stamp (start or end time depending on {@link #isStart}).
+ */
public long getEventTime() {
if (isStart) {
return startTime.getMillis();
@@ -188,6 +204,9 @@
}
}
+ /**
+ * Gets the distance from the ride location to the given one.
+ */
public double getEuclideanDistance(double longitude, double latitude) {
if (this.isStart) {
return GeoUtils.getEuclideanDistance((float) longitude, (float) latitude, this.startLon, this.startLat);
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 63b77e7..c0f3c88 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
@@ -39,19 +39,20 @@
* read from a gzipped input file. Each record has a time stamp and the input file must be
* ordered by this time stamp.
*
- * In order to simulate a realistic stream source, the SourceFunction serves events proportional to
+ * <p>In order to simulate a realistic stream source, the SourceFunction serves events proportional to
* their timestamps. In addition, the serving of events can be delayed by a bounded random delay
* which causes the events to be served slightly out-of-order of their timestamps.
*
- * The serving speed of the SourceFunction can be adjusted by a serving speed factor.
+ * <p>The serving speed of the SourceFunction can be adjusted by a serving speed factor.
* A factor of 60.0 increases the logical serving time by a factor of 60, i.e., events of one
* minute (60 seconds) are served in 1 second.
*
- * This SourceFunction is an EventSourceFunction and does continuously emit watermarks.
+ * <p>This SourceFunction is an EventSourceFunction and does continuously emit watermarks.
* Hence it is able to operate in event time mode which is configured as follows:
*
- * StreamExecutionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
- *
+ * <code>
+ * StreamExecutionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ * </code>
*/
public class TaxiFareSource implements SourceFunction<TaxiFare> {
@@ -97,7 +98,7 @@
* @param servingSpeedFactor The serving speed factor by which the logical serving time is adjusted.
*/
public TaxiFareSource(String dataFilePath, int maxEventDelaySecs, int servingSpeedFactor) {
- if(maxEventDelaySecs < 0) {
+ if (maxEventDelaySecs < 0) {
throw new IllegalArgumentException("Max event delay must be positive");
}
this.dataFilePath = dataFilePath;
@@ -168,12 +169,11 @@
// insert all events into schedule that might be emitted next
long curNextDelayedEventTime = !emitSchedule.isEmpty() ? emitSchedule.peek().f0 : -1;
long rideEventTime = fare != null ? getEventTime(fare) : -1;
- while(
- fare != null && ( // while there is a ride AND
+ while (
+ fare != null && (// while there is a ride AND
emitSchedule.isEmpty() || // and no ride in schedule OR
rideEventTime < curNextDelayedEventTime + maxDelayMsecs) // not enough rides in schedule
- )
- {
+ ) {
// insert event into emit schedule
long delayedEventTime = rideEventTime + getNormalDelayMsecs(rand);
emitSchedule.add(new Tuple2<Long, Object>(delayedEventTime, fare));
@@ -197,15 +197,15 @@
long servingTime = toServingTime(servingStartTime, dataStartTime, delayedEventTime);
long waitTime = servingTime - now;
- Thread.sleep( (waitTime > 0) ? waitTime : 0);
+ Thread.sleep((waitTime > 0) ? waitTime : 0);
- if(head.f1 instanceof TaxiFare) {
- TaxiFare emitFare = (TaxiFare)head.f1;
+ if (head.f1 instanceof TaxiFare) {
+ TaxiFare emitFare = (TaxiFare) head.f1;
// emit ride
sourceContext.collectWithTimestamp(emitFare, getEventTime(emitFare));
}
- else if(head.f1 instanceof Watermark) {
- Watermark emitWatermark = (Watermark)head.f1;
+ else if (head.f1 instanceof Watermark) {
+ Watermark emitWatermark = (Watermark) head.f1;
// emit watermark
sourceContext.emitWatermark(emitWatermark);
// schedule next watermark
@@ -216,20 +216,20 @@
}
}
- public long toServingTime(long servingStartTime, long dataStartTime, long eventTime) {
+ protected long toServingTime(long servingStartTime, long dataStartTime, long eventTime) {
long dataDiff = eventTime - dataStartTime;
return servingStartTime + (dataDiff / this.servingSpeed);
}
- public long getEventTime(TaxiFare fare) {
+ protected long getEventTime(TaxiFare fare) {
return fare.getEventTime();
}
- public long getNormalDelayMsecs(Random rand) {
+ protected long getNormalDelayMsecs(Random rand) {
long delay = -1;
long x = maxDelayMsecs / 2;
- while(delay < 0 || delay > maxDelayMsecs) {
- delay = (long)(rand.nextGaussian() * x) + x;
+ while (delay < 0 || delay > maxDelayMsecs) {
+ delay = (long) (rand.nextGaussian() * x) + x;
}
return delay;
}
@@ -243,7 +243,7 @@
if (this.gzipStream != null) {
this.gzipStream.close();
}
- } catch(IOException ioe) {
+ } catch (IOException ioe) {
throw new RuntimeException("Could not cancel SourceFunction", ioe);
} finally {
this.reader = null;
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 f57c41c..6bfb970 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
@@ -39,19 +39,20 @@
* read from a gzipped input file. Each record has a time stamp and the input file must be
* ordered by this time stamp.
*
- * In order to simulate a realistic stream source, the SourceFunction serves events proportional to
+ * <p>In order to simulate a realistic stream source, the SourceFunction serves events proportional to
* their timestamps. In addition, the serving of events can be delayed by a bounded random delay
* which causes the events to be served slightly out-of-order of their timestamps.
*
- * The serving speed of the SourceFunction can be adjusted by a serving speed factor.
+ * <p>The serving speed of the SourceFunction can be adjusted by a serving speed factor.
* A factor of 60.0 increases the logical serving time by a factor of 60, i.e., events of one
* minute (60 seconds) are served in 1 second.
*
- * This SourceFunction is an EventSourceFunction and does continuously emit watermarks.
+ * <p>This SourceFunction is an EventSourceFunction and does continuously emit watermarks.
* Hence it is able to operate in event time mode which is configured as follows:
*
- * StreamExecutionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
- *
+ * <code>
+ * StreamExecutionEnvironment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ * </code>
*/
public class TaxiRideSource implements SourceFunction<TaxiRide> {
@@ -97,7 +98,7 @@
* @param servingSpeedFactor The serving speed factor by which the logical serving time is adjusted.
*/
public TaxiRideSource(String dataFilePath, int maxEventDelaySecs, int servingSpeedFactor) {
- if(maxEventDelaySecs < 0) {
+ if (maxEventDelaySecs < 0) {
throw new IllegalArgumentException("Max event delay must be positive");
}
this.dataFilePath = dataFilePath;
@@ -168,12 +169,11 @@
// insert all events into schedule that might be emitted next
long curNextDelayedEventTime = !emitSchedule.isEmpty() ? emitSchedule.peek().f0 : -1;
long rideEventTime = ride != null ? getEventTime(ride) : -1;
- while(
- ride != null && ( // while there is a ride AND
+ while (
+ ride != null && (// while there is a ride AND
emitSchedule.isEmpty() || // and no ride in schedule OR
rideEventTime < curNextDelayedEventTime + maxDelayMsecs) // not enough rides in schedule
- )
- {
+ ) {
// insert event into emit schedule
long delayedEventTime = rideEventTime + getNormalDelayMsecs(rand);
emitSchedule.add(new Tuple2<Long, Object>(delayedEventTime, ride));
@@ -197,15 +197,15 @@
long servingTime = toServingTime(servingStartTime, dataStartTime, delayedEventTime);
long waitTime = servingTime - now;
- Thread.sleep( (waitTime > 0) ? waitTime : 0);
+ Thread.sleep((waitTime > 0) ? waitTime : 0);
- if(head.f1 instanceof TaxiRide) {
- TaxiRide emitRide = (TaxiRide)head.f1;
+ if (head.f1 instanceof TaxiRide) {
+ TaxiRide emitRide = (TaxiRide) head.f1;
// emit ride
sourceContext.collectWithTimestamp(emitRide, getEventTime(emitRide));
}
- else if(head.f1 instanceof Watermark) {
- Watermark emitWatermark = (Watermark)head.f1;
+ else if (head.f1 instanceof Watermark) {
+ Watermark emitWatermark = (Watermark) head.f1;
// emit watermark
sourceContext.emitWatermark(emitWatermark);
// schedule next watermark
@@ -216,20 +216,20 @@
}
}
- public long toServingTime(long servingStartTime, long dataStartTime, long eventTime) {
+ protected long toServingTime(long servingStartTime, long dataStartTime, long eventTime) {
long dataDiff = eventTime - dataStartTime;
return servingStartTime + (dataDiff / this.servingSpeed);
}
- public long getEventTime(TaxiRide ride) {
+ protected long getEventTime(TaxiRide ride) {
return ride.getEventTime();
}
- public long getNormalDelayMsecs(Random rand) {
+ protected long getNormalDelayMsecs(Random rand) {
long delay = -1;
long x = maxDelayMsecs / 2;
- while(delay < 0 || delay > maxDelayMsecs) {
- delay = (long)(rand.nextGaussian() * x) + x;
+ while (delay < 0 || delay > maxDelayMsecs) {
+ delay = (long) (rand.nextGaussian() * x) + x;
}
return delay;
}
@@ -243,7 +243,7 @@
if (this.gzipStream != null) {
this.gzipStream.close();
}
- } catch(IOException ioe) {
+ } catch (IOException ioe) {
throw new RuntimeException("Could not cancel SourceFunction", ioe);
} finally {
this.reader = null;
diff --git a/common/src/main/java/org/apache/flink/training/exercises/common/utils/ExerciseBase.java b/common/src/main/java/org/apache/flink/training/exercises/common/utils/ExerciseBase.java
index b910710..3440a15 100644
--- a/common/src/main/java/org/apache/flink/training/exercises/common/utils/ExerciseBase.java
+++ b/common/src/main/java/org/apache/flink/training/exercises/common/utils/ExerciseBase.java
@@ -23,6 +23,9 @@
import org.apache.flink.training.exercises.common.datatypes.TaxiFare;
import org.apache.flink.training.exercises.common.datatypes.TaxiRide;
+/**
+ * Base for all exercises with a few helper methods.
+ */
public class ExerciseBase {
public static SourceFunction<TaxiRide> rides = null;
public static SourceFunction<TaxiFare> fares = null;
@@ -30,9 +33,12 @@
public static SinkFunction out = null;
public static int parallelism = 4;
- public final static String pathToRideData = "/Users/david/stuff/flink-training/trainingData/nycTaxiRides.gz";
- public final static String pathToFareData = "/Users/david/stuff/flink-training/trainingData/nycTaxiFares.gz";
+ public static final String PATH_TO_RIDE_DATA = "/Users/david/stuff/flink-training/trainingData/nycTaxiRides.gz";
+ public static final String PATH_TO_FARE_DATA = "/Users/david/stuff/flink-training/trainingData/nycTaxiFares.gz";
+ /**
+ * Retrieves a test source during unit tests and the given one during normal execution.
+ */
public static SourceFunction<TaxiRide> rideSourceOrTest(SourceFunction<TaxiRide> source) {
if (rides == null) {
return source;
@@ -40,6 +46,9 @@
return rides;
}
+ /**
+ * Retrieves a test source during unit tests and the given one during normal execution.
+ */
public static SourceFunction<TaxiFare> fareSourceOrTest(SourceFunction<TaxiFare> source) {
if (fares == null) {
return source;
@@ -47,6 +56,9 @@
return fares;
}
+ /**
+ * Retrieves a test source during unit tests and the given one during normal execution.
+ */
public static SourceFunction<String> stringSourceOrTest(SourceFunction<String> source) {
if (strings == null) {
return source;
@@ -54,6 +66,9 @@
return strings;
}
+ /**
+ * Prints the given data stream during normal execution and collects outputs during tests.
+ */
public static void printOrTest(org.apache.flink.streaming.api.datastream.DataStream<?> ds) {
if (out == null) {
ds.print();
@@ -62,6 +77,9 @@
}
}
+ /**
+ * Prints the given data stream during normal execution and collects outputs during tests.
+ */
public static void printOrTest(org.apache.flink.streaming.api.scala.DataStream<?> ds) {
if (out == null) {
ds.print();
@@ -69,4 +87,4 @@
ds.addSink(out);
}
}
-}
\ No newline at end of file
+}
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 9f5d084..4594248 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
@@ -28,25 +28,25 @@
public class GeoUtils {
// geo boundaries of the area of NYC
- public static double LON_EAST = -73.7;
- public static double LON_WEST = -74.05;
- public static double LAT_NORTH = 41.0;
- public static double LAT_SOUTH = 40.5;
+ public static final double LON_EAST = -73.7;
+ public static final double LON_WEST = -74.05;
+ public static final double LAT_NORTH = 41.0;
+ public static final double LAT_SOUTH = 40.5;
// area width and height
- public static double LON_WIDTH = 74.05 - 73.7;
- public static double LAT_HEIGHT = 41.0 - 40.5;
+ public static final double LON_WIDTH = 74.05 - 73.7;
+ public static final double LAT_HEIGHT = 41.0 - 40.5;
// delta step to create artificial grid overlay of NYC
- public static double DELTA_LON = 0.0014;
- public static double DELTA_LAT = 0.00125;
+ public static final double DELTA_LON = 0.0014;
+ public static final double DELTA_LAT = 0.00125;
// ( |LON_WEST| - |LON_EAST| ) / DELTA_LAT
- public static int NUMBER_OF_GRID_X = 250;
+ public static final int NUMBER_OF_GRID_X = 250;
// ( LAT_NORTH - LAT_SOUTH ) / DELTA_LON
- public static int NUMBER_OF_GRID_Y = 400;
+ public static final int NUMBER_OF_GRID_Y = 400;
- public static float DEG_LEN = 110.25f;
+ public static final float DEG_LEN = 110.25f;
/**
* Checks if a location specified by longitude and latitude values is
@@ -103,7 +103,7 @@
int y2 = (int) Math.floor((LAT_NORTH - lat2) / DELTA_LAT);
int startX, startY, endX, endY;
- if(x1 <= x2) {
+ if (x1 <= x2) {
startX = x1;
startY = y1;
endX = x2;
@@ -116,7 +116,7 @@
endY = y1;
}
- double slope = (endY - startY) / ((endX - startX)+0.00000001);
+ double slope = (endY - startY) / ((endX - startX) + 0.00000001);
int curX = startX;
int curY = startY;
@@ -124,16 +124,16 @@
ArrayList<Integer> cellIds = new ArrayList<>(64);
cellIds.add(curX + (curY * NUMBER_OF_GRID_X));
- while(curX < endX || curY != endY) {
+ while (curX < endX || curY != endY) {
- if(slope > 0) {
+ if (slope > 0) {
double y = (curX - startX + 0.5) * slope + startY - 0.5;
- if(y > curY - 0.05 && y < curY + 0.05) {
+ if (y > curY - 0.05 && y < curY + 0.05) {
curX++;
curY++;
}
- else if(y < curY) {
+ else if (y < curY) {
curX++;
}
else {
@@ -143,11 +143,11 @@
else {
double y = (curX - startX + 0.5) * slope + startY + 0.5;
- if(y > curY - 0.05 && y < curY + 0.05) {
+ if (y > curY - 0.05 && y < curY + 0.05) {
curX++;
curY--;
}
- if(y > curY) {
+ if (y > curY) {
curX++;
}
else {
@@ -173,7 +173,7 @@
int xIndex = gridCellId % NUMBER_OF_GRID_X;
- return (float)(Math.abs(LON_WEST) - (xIndex * DELTA_LON) - (DELTA_LON / 2)) * -1.0f;
+ return (float) (Math.abs(LON_WEST) - (xIndex * DELTA_LON) - (DELTA_LON / 2)) * -1.0f;
}
/**
@@ -188,7 +188,7 @@
int xIndex = gridCellId % NUMBER_OF_GRID_X;
int yIndex = (gridCellId - xIndex) / NUMBER_OF_GRID_X;
- return (float)(LAT_NORTH - (yIndex * DELTA_LAT) - (DELTA_LAT / 2));
+ return (float) (LAT_NORTH - (yIndex * DELTA_LAT) - (DELTA_LAT / 2));
}
@@ -199,7 +199,7 @@
* @return A random longitude value within the NYC area.
*/
public static float getRandomNYCLon(Random rand) {
- return (float)(LON_EAST - (LON_WIDTH * rand.nextFloat()));
+ return (float) (LON_EAST - (LON_WIDTH * rand.nextFloat()));
}
/**
@@ -209,7 +209,7 @@
* @return A random latitude value within the NYC area.
*/
public static float getRandomNYCLat(Random rand) {
- return (float)(LAT_SOUTH + (LAT_HEIGHT * rand.nextFloat()));
+ return (float) (LAT_SOUTH + (LAT_HEIGHT * rand.nextFloat()));
}
/**
@@ -224,14 +224,14 @@
public static double getEuclideanDistance(float lon1, float lat1, float lon2, float lat2) {
double x = lat1 - lat2;
double y = (lon1 - lon2) * Math.cos(lat2);
- return (DEG_LEN * Math.sqrt(x*x + y*y));
+ return (DEG_LEN * Math.sqrt(x * x + y * y));
}
/**
* Returns the angle in degrees between the vector from the start to the destination
* and the x-axis on which the start is located.
*
- * The angle describes in which direction the destination is located from the start, i.e.,
+ * <p>The angle describes in which direction the destination is located from the start, i.e.,
* 0° -> East, 90° -> South, 180° -> West, 270° -> North
*
* @param startLon longitude of start location
diff --git a/common/src/main/java/org/apache/flink/training/exercises/common/utils/MissingSolutionException.java b/common/src/main/java/org/apache/flink/training/exercises/common/utils/MissingSolutionException.java
index 7d9a977..4f6c95f 100644
--- a/common/src/main/java/org/apache/flink/training/exercises/common/utils/MissingSolutionException.java
+++ b/common/src/main/java/org/apache/flink/training/exercises/common/utils/MissingSolutionException.java
@@ -18,18 +18,12 @@
package org.apache.flink.training.exercises.common.utils;
+/**
+ * Exception denoting a missing solution (results in tests verifying the solution instead).
+ */
public class MissingSolutionException extends Exception {
- public MissingSolutionException() {};
-
- public MissingSolutionException(String message) {
- super(message);
- };
-
- public MissingSolutionException(Throwable cause) {
- super(cause);
- }
-
- public MissingSolutionException(String message, Throwable cause) {
- super(message, cause);
- }
-};
+ /**
+ * Create new exception.
+ */
+ public MissingSolutionException() {}
+}
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 c5b6f95..6c59de1 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
@@ -66,11 +66,11 @@
public static class TestSink<OUT> implements SinkFunction<OUT> {
// must be static
- public static final List values = new ArrayList<>();
+ public static final List VALUES = new ArrayList<>();
@Override
public void invoke(OUT value, Context context) throws Exception {
- values.add(value);
+ VALUES.add(value);
}
}
@@ -111,7 +111,7 @@
}
private List<OUT> execute(TestSink<OUT> sink, Testable exercise, Testable solution) throws Exception {
- sink.values.clear();
+ sink.VALUES.clear();
ExerciseBase.out = sink;
ExerciseBase.parallelism = 1;
@@ -120,25 +120,25 @@
exercise.main();
} catch (Exception e) {
if (ultimateCauseIsMissingSolution(e)) {
- sink.values.clear();
+ sink.VALUES.clear();
solution.main();
} else {
throw e;
}
}
- return sink.values;
+ return sink.VALUES;
}
private List<OUT> execute(TestSink<OUT> sink, Testable solution) throws Exception {
- sink.values.clear();
+ sink.VALUES.clear();
ExerciseBase.out = sink;
ExerciseBase.parallelism = 1;
solution.main();
- return sink.values;
+ return sink.VALUES;
}
private boolean ultimateCauseIsMissingSolution(Throwable e) {
diff --git a/config/checkstyle/checkstyle.xml b/config/checkstyle/checkstyle.xml
new file mode 100644
index 0000000..34b0e76
--- /dev/null
+++ b/config/checkstyle/checkstyle.xml
@@ -0,0 +1,570 @@
+<?xml version="1.0"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+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">
+
+<!--
+This is a checkstyle configuration file. For descriptions of
+what the following rules do, please see the checkstyle configuration
+page at http://checkstyle.sourceforge.net/config.html.
+
+This file is based on the checkstyle file of Apache Beam.
+-->
+
+<module name="Checker">
+
+ <!--<module name="FileTabCharacter">-->
+ <!--<!– Checks that there are no tab characters in the file. –>-->
+ <!--</module>-->
+
+ <module name="NewlineAtEndOfFile">
+ <!-- windows can use \r\n vs \n, so enforce the most used one ie UNIx style -->
+ <property name="lineSeparator" value="lf" />
+ </module>
+
+ <module name="RegexpSingleline">
+ <!-- Checks that TODOs don't have stuff in parenthesis, e.g., username. -->
+ <property name="format" value="((//.*)|(\*.*))TODO\(" />
+ <property name="message" value="TODO comments must not include usernames." />
+ <property name="severity" value="error" />
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format" value="\s+$"/>
+ <property name="message" value="Trailing whitespace"/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="RegexpSingleline">
+ <property name="format" value="Throwables.propagate\("/>
+ <property name="message" value="Throwables.propagate is deprecated"/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <!-- Prevent *Tests.java as tools may not pick them up -->
+ <!--<module name="RegexpOnFilename">-->
+ <!--<property name="fileNamePattern" value=".*Tests\.java$" />-->
+ <!--</module>-->
+
+ <module name="SuppressionFilter">
+ <property name="file" value="${config_loc}/suppressions.xml" default="suppressions.xml" />
+ </module>
+
+ <!-- Check that every module has a package-info.java -->
+ <!--<module name="JavadocPackage"/>-->
+
+ <!--
+
+ FLINK CUSTOM CHECKS
+
+ -->
+
+ <module name="FileLength">
+ <property name="max" value="3000"/>
+ </module>
+
+ <!-- All Java AST specific tests live under TreeWalker module. -->
+ <module name="TreeWalker">
+
+ <!-- Allow use of comment to suppress javadocstyle -->
+ <module name="SuppressionCommentFilter">
+ <property name="offCommentFormat" value="CHECKSTYLE.OFF\: ([\w\|]+)"/>
+ <property name="onCommentFormat" value="CHECKSTYLE.ON\: ([\w\|]+)"/>
+ <property name="checkFormat" value="$1"/>
+ </module>
+
+ <!--
+
+ FLINK CUSTOM CHECKS
+
+ -->
+
+ <module name="RegexpSinglelineJava">
+ <property name="format" value="^\t* +\t*\S" />
+ <property name="message"
+ value="Line has leading space characters; indentation should be performed with tabs only." />
+ <property name="ignoreComments" value="true" />
+ </module>
+
+ <!-- Prohibit T.getT() methods for standard boxed types -->
+ <module name="Regexp">
+ <property name="format" value="Boolean\.getBoolean"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use System.getProperties() to get system properties."/>
+ </module>
+
+ <module name="Regexp">
+ <property name="format" value="Integer\.getInteger"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use System.getProperties() to get system properties."/>
+ </module>
+
+ <module name="Regexp">
+ <property name="format" value="Long\.getLong"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use System.getProperties() to get system properties."/>
+ </module>
+
+ <!--
+
+ IllegalImport cannot blacklist classes so we have to fall back to Regexp.
+
+ -->
+
+ <!-- forbid use of commons lang validate -->
+ <module name="Regexp">
+ <property name="format" value="org\.apache\.commons\.lang3\.Validate"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use Guava Checks instead of Commons Validate. Please refer to the coding guidelines."/>
+ </module>
+ <!-- forbid the use of google.common.base.Preconditions -->
+ <module name="Regexp">
+ <property name="format" value="import com\.google\.common\.base\.Preconditions"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use Flink's Preconditions instead of Guava's Preconditions"/>
+ </module>
+ <!-- forbid the use of com.google.common.annotations.VisibleForTesting -->
+ <module name="Regexp">
+ <property name="format" value="import com\.google\.common\.annotations\.VisibleForTesting"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use Flink's VisibleForTesting instead of Guava's VisibleForTesting"/>
+ </module>
+ <module name="Regexp">
+ <property name="format" value="import static com\.google\.common\.base\.Preconditions"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use Flink's Preconditions instead of Guava's Preconditions"/>
+ </module>
+ <!-- forbid the use of org.apache.commons.lang.SerializationUtils -->
+ <module name="Regexp">
+ <property name="format" value="org\.apache\.commons\.lang\.SerializationUtils"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use Flink's InstantiationUtil instead of common's SerializationUtils"/>
+ </module>
+ <module name="Regexp">
+ <property name="format" value="org\.apache\.commons\.lang3\.SerializationUtils"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use Flink's InstantiationUtil instead of common's SerializationUtils"/>
+ </module>
+ <module name="Regexp">
+ <property name="format" value="org\.apache\.commons\.lang\."/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use commons-lang3 instead of commons-lang."/>
+ </module>
+ <module name="Regexp">
+ <property name="format" value="org\.codehaus\.jettison"/>
+ <property name="illegalPattern" value="true"/>
+ <property name="message" value="Use com.fasterxml.jackson instead of jettison."/>
+ </module>
+
+ <!-- Enforce Java-style array declarations -->
+ <module name="ArrayTypeStyle" />
+
+ <module name="TodoComment">
+ <!-- Checks that disallowed strings are not used in comments. -->
+ <property name="format" value="(FIXME)|(XXX)|(@author)" />
+ </module>
+
+ <!--
+
+ IMPORT CHECKS
+
+ -->
+
+ <module name="RedundantImport">
+ <!-- Checks for redundant import statements. -->
+ <property name="severity" value="error"/>
+ <message key="import.redundancy"
+ value="Redundant import {0}."/>
+ </module>
+
+ <module name="ImportOrder">
+ <!-- Checks for out of order import statements. -->
+ <property name="severity" value="error"/>
+ <property name="groups" value="org.apache.flink,org.apache.flink.shaded,*,javax,java,scala"/>
+ <property name="separated" value="true"/>
+ <property name="sortStaticImportsAlphabetically" value="true"/>
+ <property name="option" value="bottom"/>
+ <property name="tokens" value="STATIC_IMPORT, IMPORT"/>
+ <message key="import.ordering"
+ value="Import {0} appears after other imports that it should precede"/>
+ </module>
+
+ <module name="AvoidStarImport">
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="IllegalImport">
+ <property name="illegalPkgs" value="autovalue.shaded, avro.shaded, com.google.api.client.repackaged, com.google.appengine.repackaged, org.codehaus.jackson, io.netty, org.objectweb.asm, com.google.common"/>
+ </module>
+
+ <module name="RedundantModifier">
+ <!-- Checks for redundant modifiers on various symbol definitions.
+ See: http://checkstyle.sourceforge.net/config_modifier.html#RedundantModifier
+
+ We exclude METHOD_DEF to allow final methods in final classes to make them more future-proof.
+ -->
+ <property name="tokens" value="VARIABLE_DEF, ANNOTATION_FIELD_DEF, INTERFACE_DEF, CLASS_DEF, ENUM_DEF"/>
+ </module>
+
+ <!--
+ IllegalImport cannot blacklist classes, and c.g.api.client.util is used for some shaded
+ code and some useful code. So we need to fall back to Regexp.
+ -->
+ <!--<module name="RegexpSinglelineJava">-->
+ <!--<property name="format" value="com\.google\.api\.client\.util\.(ByteStreams|Charsets|Collections2|Joiner|Lists|Maps|Objects|Preconditions|Sets|Strings|Throwables)"/>-->
+ <!--</module>-->
+
+ <!--
+ Require static importing from Preconditions.
+ -->
+ <module name="RegexpSinglelineJava">
+ <property name="format" value="^import com.google.common.base.Preconditions;$"/>
+ <property name="message" value="Static import functions from Guava Preconditions"/>
+ </module>
+
+ <module name="UnusedImports">
+ <property name="severity" value="error"/>
+ <property name="processJavadoc" value="true"/>
+ <message key="import.unused"
+ value="Unused import: {0}."/>
+ </module>
+
+ <!--
+
+ JAVADOC CHECKS
+
+ -->
+
+ <!-- Checks for Javadoc comments. -->
+ <!-- See http://checkstyle.sf.net/config_javadoc.html -->
+ <module name="JavadocMethod">
+ <property name="scope" value="protected"/>
+ <property name="severity" value="error"/>
+ <property name="allowMissingParamTags" value="true"/>
+ <property name="allowMissingReturnTag" value="true"/>
+ <property name="validateThrows" value="false"/>
+ </module>
+
+ <module name="MissingJavadocMethod">
+ </module>
+
+ <!-- Check that paragraph tags are used correctly in Javadoc. -->
+ <module name="JavadocParagraph"/>
+
+ <module name="JavadocType">
+ <property name="scope" value="protected"/>
+ <property name="severity" value="error"/>
+ <property name="allowMissingParamTags" value="true"/>
+ </module>
+
+ <module name="JavadocStyle">
+ <property name="severity" value="error"/>
+ <property name="checkHtml" value="true"/>
+ </module>
+
+ <!--
+
+ NAMING CHECKS
+
+ -->
+
+ <!-- Item 38 - Adhere to generally accepted naming conventions -->
+
+ <module name="PackageName">
+ <!-- Validates identifiers for package names against the
+ supplied expression. -->
+ <!-- Here the default checkstyle rule restricts package name parts to
+ seven characters, this is not in line with common practice at Google.
+ -->
+ <property name="format" value="^[a-z]+(\.[a-z][a-z0-9]{1,})*$"/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="TypeNameCheck">
+ <!-- Validates static, final fields against the
+ expression "^[A-Z][a-zA-Z0-9]*$". -->
+ <metadata name="altname" value="TypeName"/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="ConstantNameCheck">
+ <!-- Validates non-private, static, final fields against the supplied
+ public/package final fields "^[A-Z][A-Z0-9]*(_[A-Z0-9]+)*$". -->
+ <metadata name="altname" value="ConstantName"/>
+ <property name="applyToPublic" value="true"/>
+ <property name="applyToProtected" value="true"/>
+ <property name="applyToPackage" value="true"/>
+ <property name="applyToPrivate" value="false"/>
+ <property name="format" value="^([A-Z][A-Z0-9]*(_[A-Z0-9]+)*|FLAG_.*)$"/>
+ <message key="name.invalidPattern"
+ value="Variable ''{0}'' should be in ALL_CAPS (if it is a constant) or be private (otherwise)."/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="StaticVariableNameCheck">
+ <!-- Validates static, non-final fields against the supplied
+ expression "^[a-z][a-zA-Z0-9]*_?$". -->
+ <metadata name="altname" value="StaticVariableName"/>
+ <property name="applyToPublic" value="true"/>
+ <property name="applyToProtected" value="true"/>
+ <property name="applyToPackage" value="true"/>
+ <property name="applyToPrivate" value="true"/>
+ <property name="format" value="^[a-z][a-zA-Z0-9]*_?$"/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="MemberNameCheck">
+ <!-- Validates non-static members against the supplied expression. -->
+ <metadata name="altname" value="MemberName"/>
+ <property name="applyToPublic" value="true"/>
+ <property name="applyToProtected" value="true"/>
+ <property name="applyToPackage" value="true"/>
+ <property name="applyToPrivate" value="true"/>
+ <property name="format" value="^[a-z][a-zA-Z0-9]*$"/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="MethodNameCheck">
+ <!-- Validates identifiers for method names. -->
+ <metadata name="altname" value="MethodName"/>
+ <property name="format" value="^[a-z][a-zA-Z0-9]*(_[a-zA-Z0-9]+)*$"/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="ParameterName">
+ <!-- Validates identifiers for method parameters against the
+ expression "^[a-z][a-zA-Z0-9]*$". -->
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="LocalFinalVariableName">
+ <!-- Validates identifiers for local final variables against the
+ expression "^[a-z][a-zA-Z0-9]*$". -->
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="LocalVariableName">
+ <!-- Validates identifiers for local variables against the
+ expression "^[a-z][a-zA-Z0-9]*$". -->
+ <property name="severity" value="error"/>
+ </module>
+
+ <!-- Type parameters must be either one of the four blessed letters
+ T, K, V, W, X or else be capital-case terminated with a T,
+ such as MyGenericParameterT -->
+ <!--<module name="ClassTypeParameterName">-->
+ <!--<property name="format" value="^(((T|K|V|W|X)[0-9]*)|([A-Z][a-z][a-zA-Z]*T))$"/>-->
+ <!--<property name="severity" value="error"/>-->
+ <!--</module>-->
+
+ <!--<module name="MethodTypeParameterName">-->
+ <!--<property name="format" value="^(((T|K|V|W|X)[0-9]*)|([A-Z][a-z][a-zA-Z]*T))$"/>-->
+ <!--<property name="severity" value="error"/>-->
+ <!--</module>-->
+
+ <!--<module name="InterfaceTypeParameterName">-->
+ <!--<property name="format" value="^(((T|K|V|W|X)[0-9]*)|([A-Z][a-z][a-zA-Z]*T))$"/>-->
+ <!--<property name="severity" value="error"/>-->
+ <!--</module>-->
+
+ <!--
+
+ LENGTH and CODING CHECKS
+
+ -->
+
+ <!--<module name="LineLength">-->
+ <!--<!– Checks if a line is too long. –>-->
+ <!--<property name="max" value="100"/>-->
+ <!--<property name="severity" value="error"/>-->
+
+ <!--<!–-->
+ <!--The default ignore pattern exempts the following elements:-->
+ <!-- - import statements-->
+ <!-- - long URLs inside comments-->
+ <!--–>-->
+
+ <!--<property name="ignorePattern"-->
+ <!--value="^(package .*;\s*)|(import .*;\s*)|( *\* .*https?://.*)$"/>-->
+ <!--</module>-->
+
+ <module name="LeftCurly">
+ <!-- Checks for placement of the left curly brace ('{'). -->
+ <property name="severity" value="error"/>
+ </module>
+
+ <!--<module name="RightCurly">-->
+ <!--<!– Checks right curlies on CATCH, ELSE, and TRY blocks are on-->
+ <!--the same line. e.g., the following example is fine:-->
+ <!--<pre>-->
+ <!--if {-->
+ <!--...-->
+ <!--} else-->
+ <!--</pre>-->
+ <!--–>-->
+ <!--<!– This next example is not fine:-->
+ <!--<pre>-->
+ <!--if {-->
+ <!--...-->
+ <!--}-->
+ <!--else-->
+ <!--</pre>-->
+ <!--–>-->
+ <!--<property name="option" value="same"/>-->
+ <!--<property name="severity" value="error"/>-->
+ <!--</module>-->
+
+ <!-- Checks for braces around if and else blocks -->
+ <module name="NeedBraces">
+ <property name="severity" value="error"/>
+ <property name="tokens" value="LITERAL_IF, LITERAL_ELSE, LITERAL_FOR, LITERAL_WHILE, LITERAL_DO"/>
+ </module>
+
+ <module name="UpperEll">
+ <!-- Checks that long constants are defined with an upper ell.-->
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="FallThrough">
+ <!-- Warn about falling through to the next case statement. Similar to
+ javac -Xlint:fallthrough, but the check is suppressed if a single-line comment
+ on the last non-blank line preceding the fallen-into case contains 'fall through' (or
+ some other variants that we don't publicized to promote consistency).
+ -->
+ <property name="reliefPattern"
+ value="fall through|Fall through|fallthru|Fallthru|falls through|Falls through|fallthrough|Fallthrough|No break|NO break|no break|continue on"/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <!-- Checks for over-complicated boolean expressions. -->
+ <module name="SimplifyBooleanExpression"/>
+
+ <!-- Detects empty statements (standalone ";" semicolon). -->
+ <module name="EmptyStatement"/>
+
+ <!-- Detect multiple consecutive semicolons (e.g. ";;"). -->
+ <module name="RegexpSinglelineJava">
+ <property name="format" value=";{2,}"/>
+ <property name="message" value="Use one semicolon"/>
+ <property name="ignoreComments" value="true"/>
+ </module>
+
+ <!--
+
+ MODIFIERS CHECKS
+
+ -->
+
+ <module name="ModifierOrder">
+ <!-- Warn if modifier order is inconsistent with JLS3 8.1.1, 8.3.1, and
+ 8.4.3. The prescribed order is:
+ public, protected, private, abstract, static, final, transient, volatile,
+ synchronized, native, strictfp
+ -->
+ <property name="severity" value="error"/>
+ </module>
+
+
+ <!--
+
+ WHITESPACE CHECKS
+
+ -->
+
+ <module name="EmptyLineSeparator">
+ <!-- Checks for empty line separator between tokens. The only
+ excluded token is VARIABLE_DEF, allowing class fields to
+ be declared on consecutive lines.
+ -->
+ <property name="allowMultipleEmptyLines" value="false"/>
+ <property name="allowMultipleEmptyLinesInsideClassMembers" value="false"/>
+ <property name="tokens" value="PACKAGE_DEF, IMPORT, STATIC_IMPORT, CLASS_DEF,
+ INTERFACE_DEF, ENUM_DEF, STATIC_INIT, INSTANCE_INIT, METHOD_DEF,
+ CTOR_DEF"/>
+ </module>
+
+ <module name="WhitespaceAround">
+ <!-- Checks that various tokens are surrounded by whitespace.
+ This includes most binary operators and keywords followed
+ by regular or curly braces.
+ -->
+ <property name="tokens" value="ASSIGN, BAND, BAND_ASSIGN, BOR,
+ BOR_ASSIGN, BSR, BSR_ASSIGN, BXOR, BXOR_ASSIGN, COLON, DIV, DIV_ASSIGN,
+ EQUAL, GE, GT, LAMBDA, LAND, LE, LITERAL_CATCH, LITERAL_DO, LITERAL_ELSE,
+ LITERAL_FINALLY, LITERAL_FOR, LITERAL_IF, LITERAL_RETURN,
+ LITERAL_SYNCHRONIZED, LITERAL_TRY, LITERAL_WHILE, LOR, LT, MINUS,
+ MINUS_ASSIGN, MOD, MOD_ASSIGN, NOT_EQUAL, PLUS, PLUS_ASSIGN, QUESTION,
+ SL, SL_ASSIGN, SR_ASSIGN, STAR, STAR_ASSIGN, TYPE_EXTENSION_AND"/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="WhitespaceAfter">
+ <!-- Checks that commas, semicolons and typecasts are followed by
+ whitespace.
+ -->
+ <property name="tokens" value="COMMA, SEMI, TYPECAST"/>
+ </module>
+
+ <module name="NoWhitespaceAfter">
+ <!-- Checks that there is no whitespace after various unary operators.
+ Linebreaks are allowed.
+ -->
+ <property name="tokens" value="BNOT, DEC, DOT, INC, LNOT, UNARY_MINUS,
+ UNARY_PLUS"/>
+ <property name="allowLineBreaks" value="true"/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <module name="NoWhitespaceBefore">
+ <!-- Checks that there is no whitespace before various unary operators.
+ Linebreaks are allowed.
+ -->
+ <property name="tokens" value="SEMI, DOT, POST_DEC, POST_INC"/>
+ <property name="allowLineBreaks" value="true"/>
+ <property name="severity" value="error"/>
+ </module>
+
+ <!--<module name="OperatorWrap">-->
+ <!--<!– Checks that operators like + and ? appear at newlines rather than-->
+ <!--at the end of the previous line.-->
+ <!--–>-->
+ <!--<property name="option" value="NL"/>-->
+ <!--<property name="tokens" value="BAND, BOR, BSR, BXOR, DIV, EQUAL,-->
+ <!--GE, GT, LAND, LE, LITERAL_INSTANCEOF, LOR, LT, MINUS, MOD,-->
+ <!--NOT_EQUAL, PLUS, QUESTION, SL, SR, STAR "/>-->
+ <!--</module>-->
+
+ <module name="OperatorWrap">
+ <!-- Checks that assignment operators are at the end of the line. -->
+ <property name="option" value="eol"/>
+ <property name="tokens" value="ASSIGN"/>
+ </module>
+
+ <module name="ParenPad">
+ <!-- Checks that there is no whitespace before close parens or after
+ open parens.
+ -->
+ <property name="severity" value="error"/>
+ </module>
+
+ </module>
+</module>
+
diff --git a/config/checkstyle/suppressions.xml b/config/checkstyle/suppressions.xml
new file mode 100644
index 0000000..0b05a64
--- /dev/null
+++ b/config/checkstyle/suppressions.xml
@@ -0,0 +1,29 @@
+<?xml version="1.0"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+<!DOCTYPE suppressions PUBLIC
+ "-//Puppy Crawl//DTD Suppressions 1.1//EN"
+ "http://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
+
+<suppressions>
+ <suppress
+ files="(.*)test[/\\](.*)"
+ checks="MissingJavadocMethod"/>
+</suppressions>
diff --git a/hourly-tips/src/main/java/org/apache/flink/training/exercises/hourlytips/HourlyTipsExercise.java b/hourly-tips/src/main/java/org/apache/flink/training/exercises/hourlytips/HourlyTipsExercise.java
index ca2d5ec..b926de1 100644
--- a/hourly-tips/src/main/java/org/apache/flink/training/exercises/hourlytips/HourlyTipsExercise.java
+++ b/hourly-tips/src/main/java/org/apache/flink/training/exercises/hourlytips/HourlyTipsExercise.java
@@ -30,20 +30,27 @@
/**
* The "Hourly Tips" exercise of the Flink training in the docs.
*
- * The task of the exercise is to first calculate the total tips collected by each driver, hour by hour, and
+ * <p>The task of the exercise is to first calculate the total tips collected by each driver, hour by hour, and
* then from that stream, find the highest tip total in each hour.
*
- * Parameters:
+ * <p>Parameters:
* -input path-to-input-file
- *
*/
public class HourlyTipsExercise extends ExerciseBase {
+ /**
+ * Main method.
+ *
+ * <p>Parameters:
+ * -input path-to-input-file
+ *
+ * @throws Exception which occurs during job execution.
+ */
public static void main(String[] args) throws Exception {
// read parameters
ParameterTool params = ParameterTool.fromArgs(args);
- final String input = params.get("input", ExerciseBase.pathToFareData);
+ final String input = params.get("input", ExerciseBase.PATH_TO_FARE_DATA);
final int maxEventDelay = 60; // events are out of order by max 60 seconds
final int servingSpeedFactor = 600; // events of 10 minutes are served in 1 second
diff --git a/hourly-tips/src/main/scala/org/apache/flink/training/exercises/hourlytips/scala/HourlyTipsExercise.scala b/hourly-tips/src/main/scala/org/apache/flink/training/exercises/hourlytips/scala/HourlyTipsExercise.scala
index eda02e4..ec38975 100644
--- a/hourly-tips/src/main/scala/org/apache/flink/training/exercises/hourlytips/scala/HourlyTipsExercise.scala
+++ b/hourly-tips/src/main/scala/org/apache/flink/training/exercises/hourlytips/scala/HourlyTipsExercise.scala
@@ -33,7 +33,6 @@
*
* Parameters:
* -input path-to-input-file
- *
*/
object HourlyTipsExercise {
@@ -41,7 +40,7 @@
// read parameters
val params = ParameterTool.fromArgs(args)
- val input = params.get("input", ExerciseBase.pathToFareData)
+ val input = params.get("input", ExerciseBase.PATH_TO_FARE_DATA)
val maxDelay = 60 // events are delayed by at most 60 seconds
val speed = 600 // events of 10 minutes are served in 1 second
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 dedfd24..4e2bb32 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
@@ -34,20 +34,27 @@
/**
* Java reference implementation for the "Hourly Tips" exercise of the Flink training in the docs.
*
- * The task of the exercise is to first calculate the total tips collected by each driver, hour by hour, and
+ * <p>The task of the exercise is to first calculate the total tips collected by each driver, hour by hour, and
* then from that stream, find the highest tip total in each hour.
*
- * Parameters:
+ * <p>Parameters:
* -input path-to-input-file
- *
*/
public class HourlyTipsSolution extends ExerciseBase {
+ /**
+ * Main method.
+ *
+ * <p>Parameters:
+ * -input path-to-input-file
+ *
+ * @throws Exception which occurs during job execution.
+ */
public static void main(String[] args) throws Exception {
// read parameters
ParameterTool params = ParameterTool.fromArgs(args);
- final String input = params.get("input", ExerciseBase.pathToFareData);
+ final String input = params.get("input", ExerciseBase.PATH_TO_FARE_DATA);
final int maxEventDelay = 60; // events are out of order by max 60 seconds
final int servingSpeedFactor = 600; // events of 10 minutes are served in 1 second
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 7a32046..3ad6243 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
@@ -38,7 +38,6 @@
*
* Parameters:
* -input path-to-input-file
- *
*/
object HourlyTipsSolution {
@@ -46,7 +45,7 @@
// read parameters
val params = ParameterTool.fromArgs(args)
- val input = params.get("input", ExerciseBase.pathToFareData)
+ val input = params.get("input", ExerciseBase.PATH_TO_FARE_DATA)
val maxDelay = 60 // events are delayed by at most 60 seconds
val speed = 600 // events of 10 minutes are served in 1 second
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 2a4c70a..565a8d3 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
@@ -23,12 +23,12 @@
import org.apache.flink.training.exercises.testing.TaxiRideTestBase;
import org.apache.flink.training.solutions.hourlytips.HourlyTipsSolution;
-import com.google.common.collect.Lists;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.junit.Test;
-import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import static org.junit.Assert.assertEquals;
@@ -47,7 +47,7 @@
Tuple3<Long, Long, Float> max = new Tuple3<Long, Long, Float>(t(60), 1L, 1.0F);
- ArrayList<Tuple3<Long, Long, Float>> expected = Lists.newArrayList(max);
+ List<Tuple3<Long, Long, Float>> expected = Collections.singletonList(max);
assertEquals(expected, results(source));
}
@@ -67,7 +67,7 @@
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);
- ArrayList<Tuple3<Long, Long, Float>> expected = Lists.newArrayList(hour1, hour2);
+ List<Tuple3<Long, Long, Float>> expected = Arrays.asList(hour1, hour2);
assertEquals(expected, results(source));
}
@@ -89,7 +89,7 @@
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);
- ArrayList<Tuple3<Long, Long, Float>> expected = Lists.newArrayList(hour1, hour2);
+ List<Tuple3<Long, Long, Float>> expected = Arrays.asList(hour1, hour2);
assertEquals(expected, results(source));
}
@@ -107,4 +107,4 @@
return runApp(source, new TestSink<>(), javaExercise, javaSolution);
}
-}
\ No newline at end of file
+}
diff --git a/long-ride-alerts/src/main/java/org/apache/flink/training/exercises/longrides/LongRidesExercise.java b/long-ride-alerts/src/main/java/org/apache/flink/training/exercises/longrides/LongRidesExercise.java
index 3a291e8..327ba38 100644
--- a/long-ride-alerts/src/main/java/org/apache/flink/training/exercises/longrides/LongRidesExercise.java
+++ b/long-ride-alerts/src/main/java/org/apache/flink/training/exercises/longrides/LongRidesExercise.java
@@ -34,18 +34,26 @@
/**
* The "Long Ride Alerts" exercise of the Flink training in the docs.
*
- * The goal for this exercise is to emit START events for taxi rides that have not been matched
+ * <p>The goal for this exercise is to emit START events for taxi rides that have not been matched
* by an END event during the first 2 hours of the ride.
*
- * Parameters:
+ * <p>Parameters:
* -input path-to-input-file
- *
*/
public class LongRidesExercise extends ExerciseBase {
+
+ /**
+ * Main method.
+ *
+ * <p>Parameters:
+ * -input path-to-input-file
+ *
+ * @throws Exception which occurs during job execution.
+ */
public static void main(String[] args) throws Exception {
ParameterTool params = ParameterTool.fromArgs(args);
- final String input = params.get("input", ExerciseBase.pathToRideData);
+ final String input = params.get("input", ExerciseBase.PATH_TO_RIDE_DATA);
final int maxEventDelay = 60; // events are out of order by max 60 seconds
final int servingSpeedFactor = 600; // events of 10 minutes are served in 1 second
@@ -83,4 +91,4 @@
public void onTimer(long timestamp, OnTimerContext context, Collector<TaxiRide> out) throws Exception {
}
}
-}
\ No newline at end of file
+}
diff --git a/long-ride-alerts/src/main/scala/org/apache/flink/training/exercises/longrides/scala/LongRidesExercise.scala b/long-ride-alerts/src/main/scala/org/apache/flink/training/exercises/longrides/scala/LongRidesExercise.scala
index 966d7e4..49deb45 100644
--- a/long-ride-alerts/src/main/scala/org/apache/flink/training/exercises/longrides/scala/LongRidesExercise.scala
+++ b/long-ride-alerts/src/main/scala/org/apache/flink/training/exercises/longrides/scala/LongRidesExercise.scala
@@ -37,14 +37,14 @@
*
* Parameters:
* -input path-to-input-file
- *
*/
object LongRidesExercise {
+
def main(args: Array[String]) {
// parse parameters
val params = ParameterTool.fromArgs(args)
- val input = params.get("input", ExerciseBase.pathToRideData)
+ val input = params.get("input", ExerciseBase.PATH_TO_RIDE_DATA)
val maxDelay = 60 // events are out of order by max 60 seconds
val speed = 1800 // events of 30 minutes are served every second
diff --git a/long-ride-alerts/src/solution/java/org/apache/flink/training/solutions/longrides/LongRidesSolution.java b/long-ride-alerts/src/solution/java/org/apache/flink/training/solutions/longrides/LongRidesSolution.java
index fe47cda..fb75ef2 100644
--- a/long-ride-alerts/src/solution/java/org/apache/flink/training/solutions/longrides/LongRidesSolution.java
+++ b/long-ride-alerts/src/solution/java/org/apache/flink/training/solutions/longrides/LongRidesSolution.java
@@ -35,19 +35,26 @@
/**
* Solution to the "Long Ride Alerts" exercise of the Flink training in the docs.
*
- * The goal for this exercise is to emit START events for taxi rides that have not been matched
+ * <p>The goal for this exercise is to emit START events for taxi rides that have not been matched
* by an END event during the first 2 hours of the ride.
*
- * Parameters:
+ * <p>Parameters:
* -input path-to-input-file
- *
*/
-
public class LongRidesSolution extends ExerciseBase {
+
+ /**
+ * Main method.
+ *
+ * <p>Parameters:
+ * -input path-to-input-file
+ *
+ * @throws Exception which occurs during job execution.
+ */
public static void main(String[] args) throws Exception {
ParameterTool params = ParameterTool.fromArgs(args);
- final String input = params.get("input", ExerciseBase.pathToRideData);
+ final String input = params.get("input", ExerciseBase.PATH_TO_RIDE_DATA);
final int maxEventDelay = 60; // events are out of order by max 60 seconds
final int servingSpeedFactor = 600; // events of 10 minutes are served in 1 second
diff --git a/long-ride-alerts/src/solution/scala/org/apache/flink/training/solutions/longrides/scala/LongRidesSolution.scala b/long-ride-alerts/src/solution/scala/org/apache/flink/training/solutions/longrides/scala/LongRidesSolution.scala
index 4e182ae..29e854c 100644
--- a/long-ride-alerts/src/solution/scala/org/apache/flink/training/solutions/longrides/scala/LongRidesSolution.scala
+++ b/long-ride-alerts/src/solution/scala/org/apache/flink/training/solutions/longrides/scala/LongRidesSolution.scala
@@ -37,7 +37,6 @@
*
* Parameters:
* -input path-to-input-file
- *
*/
object LongRidesSolution {
@@ -45,7 +44,7 @@
// parse parameters
val params = ParameterTool.fromArgs(args)
- val input = params.get("input", pathToRideData)
+ val input = params.get("input", PATH_TO_RIDE_DATA)
val maxDelay = 60 // events are out of order by max 60 seconds
val speed = 1800 // events of 30 minutes are served every second
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 4000bc4..9bd68ae 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
@@ -22,10 +22,10 @@
import org.apache.flink.training.exercises.testing.TaxiRideTestBase;
import org.apache.flink.training.solutions.longrides.LongRidesSolution;
-import com.google.common.collect.Lists;
import org.joda.time.DateTime;
import org.junit.Test;
+import java.util.Collections;
import java.util.List;
import static org.junit.Assert.assertEquals;
@@ -75,7 +75,7 @@
Long markThreeHoursLater = beginning.plusHours(3).getMillis();
TestRideSource source = new TestRideSource(rideStarted, markThreeHoursLater);
- assertEquals(Lists.newArrayList(rideStarted), results(source));
+ assertEquals(Collections.singletonList(rideStarted), results(source));
}
@Test
@@ -85,11 +85,11 @@
TaxiRide rideEnded3HoursLater = endRide(rideStarted, beginning.plusHours(3));
TestRideSource source = new TestRideSource(rideStarted, mark2HoursLater, rideEnded3HoursLater);
- assertEquals(Lists.newArrayList(rideStarted), results(source));
+ assertEquals(Collections.singletonList(rideStarted), results(source));
}
private TaxiRide testRide(long rideId, Boolean isStart, DateTime startTime, DateTime endTime) {
- return new TaxiRide(rideId, isStart, startTime, endTime, -73.9947F, 40.750626F, -73.9947F, 40.750626F, (short)1, 0, 0);
+ return new TaxiRide(rideId, isStart, startTime, endTime, -73.9947F, 40.750626F, -73.9947F, 40.750626F, (short) 1, 0, 0);
}
private TaxiRide startRide(long rideId, DateTime startTime) {
@@ -105,4 +105,4 @@
return runApp(source, new TestSink<>(), javaExercise, javaSolution);
}
-}
\ No newline at end of file
+}
diff --git a/ride-cleansing/src/main/java/org/apache/flink/training/exercises/ridecleansing/RideCleansingExercise.java b/ride-cleansing/src/main/java/org/apache/flink/training/exercises/ridecleansing/RideCleansingExercise.java
index d9e1247..7e73733 100644
--- a/ride-cleansing/src/main/java/org/apache/flink/training/exercises/ridecleansing/RideCleansingExercise.java
+++ b/ride-cleansing/src/main/java/org/apache/flink/training/exercises/ridecleansing/RideCleansingExercise.java
@@ -30,18 +30,28 @@
/**
* The "Ride Cleansing" exercise from the Flink training in the docs.
*
- * The task of the exercise is to filter a data stream of taxi ride records to keep only rides that
+ * <p>The task of the exercise is to filter a data stream of taxi ride records to keep only rides that
* start and end within New York City. The resulting stream should be printed.
*
- * Parameters:
+ * <p>Parameters:
* -input path-to-input-file
- *
*/
public class RideCleansingExercise extends ExerciseBase {
+
+ /**
+ * Main method.
+ *
+ * <p>Parameters:
+ * -input path-to-input-file
+ *
+ * @throws Exception which occurs during job execution.
+ *
+ * @throws Exception which occurs during job execution.
+ */
public static void main(String[] args) throws Exception {
ParameterTool params = ParameterTool.fromArgs(args);
- final String input = params.get("input", ExerciseBase.pathToRideData);
+ final String input = params.get("input", ExerciseBase.PATH_TO_RIDE_DATA);
final int maxEventDelay = 60; // events are out of order by max 60 seconds
final int servingSpeedFactor = 600; // events of 10 minutes are served in 1 second
diff --git a/ride-cleansing/src/main/scala/org/apache/flink/training/exercises/ridecleansing/scala/RideCleansingExercise.scala b/ride-cleansing/src/main/scala/org/apache/flink/training/exercises/ridecleansing/scala/RideCleansingExercise.scala
index cc5d7c9..a0e6c6b 100644
--- a/ride-cleansing/src/main/scala/org/apache/flink/training/exercises/ridecleansing/scala/RideCleansingExercise.scala
+++ b/ride-cleansing/src/main/scala/org/apache/flink/training/exercises/ridecleansing/scala/RideCleansingExercise.scala
@@ -35,10 +35,11 @@
* -input path-to-input-file
*/
object RideCleansingExercise extends ExerciseBase {
+
def main(args: Array[String]) {
// parse parameters
val params = ParameterTool.fromArgs(args)
- val input = params.get("input", ExerciseBase.pathToRideData)
+ val input = params.get("input", ExerciseBase.PATH_TO_RIDE_DATA)
val maxDelay = 60 // events are out of order by max 60 seconds
val speed = 600 // events of 10 minutes are served in 1 second
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 1c849bd..cf3dca5 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
@@ -22,17 +22,34 @@
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-
import org.apache.flink.training.exercises.common.datatypes.TaxiRide;
import org.apache.flink.training.exercises.common.sources.TaxiRideSource;
import org.apache.flink.training.exercises.common.utils.ExerciseBase;
import org.apache.flink.training.exercises.common.utils.GeoUtils;
+/**
+ * Solution to the "Ride Cleansing" exercise of the Flink training in the docs.
+ *
+ * <p>The task of the exercise is to filter a data stream of taxi ride records to keep only rides that
+ * start and end within New York City. The resulting stream should be printed.
+ *
+ * <p>Parameters:
+ * -input path-to-input-file
+ */
public class RideCleansingSolution extends ExerciseBase {
+
+ /**
+ * Main method.
+ *
+ * <p>Parameters:
+ * -input path-to-input-file
+ *
+ * @throws Exception which occurs during job execution.
+ */
public static void main(String[] args) throws Exception {
ParameterTool params = ParameterTool.fromArgs(args);
- final String input = params.get("input", pathToRideData);
+ final String input = params.get("input", PATH_TO_RIDE_DATA);
final int maxEventDelay = 60; // events are out of order by max 60 seconds
final int servingSpeedFactor = 600; // events of 10 minutes are served in 1 second
diff --git a/ride-cleansing/src/solution/scala/org/apache/flink/training/solutions/ridecleansing/scala/RideCleansingSolution.scala b/ride-cleansing/src/solution/scala/org/apache/flink/training/solutions/ridecleansing/scala/RideCleansingSolution.scala
index 4874bbe..d831259 100644
--- a/ride-cleansing/src/solution/scala/org/apache/flink/training/solutions/ridecleansing/scala/RideCleansingSolution.scala
+++ b/ride-cleansing/src/solution/scala/org/apache/flink/training/solutions/ridecleansing/scala/RideCleansingSolution.scala
@@ -33,14 +33,13 @@
*
* Parameters:
* -input path-to-input-file
- *
*/
object RideCleansingSolution {
def main(args: Array[String]) {
// parse parameters
val params = ParameterTool.fromArgs(args)
- val input = params.get("input", pathToRideData)
+ val input = params.get("input", ExerciseBase.PATH_TO_RIDE_DATA)
val maxDelay = 60 // events are out of order by max 60 seconds
val speed = 600 // events of 10 minutes are served in 1 second
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 bce9ea0..b0e151f 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
@@ -18,7 +18,6 @@
package org.apache.flink.training.exercises.ridecleansing;
-import com.google.common.collect.Lists;
import org.apache.flink.training.exercises.common.datatypes.TaxiRide;
import org.apache.flink.training.exercises.testing.TaxiRideTestBase;
import org.apache.flink.training.solutions.ridecleansing.RideCleansingSolution;
@@ -26,6 +25,7 @@
import org.joda.time.DateTime;
import org.junit.Test;
+import java.util.Collections;
import java.util.List;
import static org.junit.Assert.assertEquals;
@@ -40,7 +40,7 @@
TestRideSource source = new TestRideSource(atPennStation);
- assertEquals(Lists.newArrayList(atPennStation), results(source));
+ assertEquals(Collections.singletonList(atPennStation), results(source));
}
@Test
@@ -51,12 +51,12 @@
TestRideSource source = new TestRideSource(toThePole, fromThePole, atNorthPole);
- assertEquals(Lists.newArrayList(), results(source));
+ assertEquals(Collections.emptyList(), results(source));
}
private TaxiRide testRide(float startLon, float startLat, float endLon, float endLat) {
return new TaxiRide(1L, true, new DateTime(0), new DateTime(0),
- startLon, startLat, endLon, endLat, (short)1, 0, 0);
+ startLon, startLat, endLon, endLat, (short) 1, 0, 0);
}
protected List<?> results(TestRideSource source) throws Exception {
@@ -64,4 +64,4 @@
return runApp(source, new TestSink<>(), javaExercise, javaSolution);
}
-}
\ No newline at end of file
+}
diff --git a/rides-and-fares/src/main/java/org/apache/flink/training/exercises/ridesandfares/RidesAndFaresExercise.java b/rides-and-fares/src/main/java/org/apache/flink/training/exercises/ridesandfares/RidesAndFaresExercise.java
index 37eac20..4bae96c 100644
--- a/rides-and-fares/src/main/java/org/apache/flink/training/exercises/ridesandfares/RidesAndFaresExercise.java
+++ b/rides-and-fares/src/main/java/org/apache/flink/training/exercises/ridesandfares/RidesAndFaresExercise.java
@@ -35,19 +35,28 @@
/**
* The "Stateful Enrichment" exercise of the Flink training in the docs.
*
- * The goal for this exercise is to enrich TaxiRides with fare information.
+ * <p>The goal for this exercise is to enrich TaxiRides with fare information.
*
- * Parameters:
+ * <p>Parameters:
* -rides path-to-input-file
* -fares path-to-input-file
- *
*/
public class RidesAndFaresExercise extends ExerciseBase {
+
+ /**
+ * Main method.
+ *
+ * <p>Parameters:
+ * -rides path-to-input-file
+ * -fares path-to-input-file
+ *
+ * @throws Exception which occurs during job execution.
+ */
public static void main(String[] args) throws Exception {
ParameterTool params = ParameterTool.fromArgs(args);
- final String ridesFile = params.get("rides", pathToRideData);
- final String faresFile = params.get("fares", pathToFareData);
+ final String ridesFile = params.get("rides", PATH_TO_RIDE_DATA);
+ final String faresFile = params.get("fares", PATH_TO_FARE_DATA);
final int delay = 60; // at most 60 seconds of delay
final int servingSpeedFactor = 1800; // 30 minutes worth of events are served every second
@@ -89,4 +98,4 @@
public void flatMap2(TaxiFare fare, Collector<Tuple2<TaxiRide, TaxiFare>> out) throws Exception {
}
}
-}
\ No newline at end of file
+}
diff --git a/rides-and-fares/src/main/scala/org/apache/flink/training/exercises/ridesandfares/scala/RidesAndFaresExercise.scala b/rides-and-fares/src/main/scala/org/apache/flink/training/exercises/ridesandfares/scala/RidesAndFaresExercise.scala
index 8b431eb..36391d8 100644
--- a/rides-and-fares/src/main/scala/org/apache/flink/training/exercises/ridesandfares/scala/RidesAndFaresExercise.scala
+++ b/rides-and-fares/src/main/scala/org/apache/flink/training/exercises/ridesandfares/scala/RidesAndFaresExercise.scala
@@ -35,15 +35,15 @@
* Parameters:
* -rides path-to-input-file
* -fares path-to-input-file
- *
*/
object RidesAndFaresExercise {
+
def main(args: Array[String]) {
// parse parameters
val params = ParameterTool.fromArgs(args)
- val ridesFile = params.get("rides", ExerciseBase.pathToRideData)
- val faresFile = params.get("fares", ExerciseBase.pathToFareData)
+ val ridesFile = params.get("rides", ExerciseBase.PATH_TO_RIDE_DATA)
+ val faresFile = params.get("fares", ExerciseBase.PATH_TO_FARE_DATA)
val delay = 60; // at most 60 seconds of delay
val servingSpeedFactor = 1800 // 30 minutes worth of events are served every second
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 7d44fd0..961aa40 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
@@ -37,18 +37,28 @@
/**
* Java reference implementation for the "Stateful Enrichment" exercise of the Flink training in the docs.
*
- * The goal for this exercise is to enrich TaxiRides with fare information.
+ * <p>The goal for this exercise is to enrich TaxiRides with fare information.
*
- * Parameters:
+ * <p>Parameters:
* -rides path-to-input-file
* -fares path-to-input-file
*/
public class RidesAndFaresSolution extends ExerciseBase {
+
+ /**
+ * Main method.
+ *
+ * <p>Parameters:
+ * -rides path-to-input-file
+ * -fares path-to-input-file
+ *
+ * @throws Exception which occurs during job execution.
+ */
public static void main(String[] args) throws Exception {
ParameterTool params = ParameterTool.fromArgs(args);
- final String ridesFile = params.get("rides", pathToRideData);
- final String faresFile = params.get("fares", pathToFareData);
+ final String ridesFile = params.get("rides", PATH_TO_RIDE_DATA);
+ final String faresFile = params.get("fares", PATH_TO_FARE_DATA);
final int delay = 60; // at most 60 seconds of delay
final int servingSpeedFactor = 1800; // 30 minutes worth of events are served every second
@@ -121,4 +131,4 @@
}
}
}
-}
\ No newline at end of file
+}
diff --git a/rides-and-fares/src/solution/scala/org/apache/flink/training/solutions/ridesandfares/scala/RidesAndFaresSolution.scala b/rides-and-fares/src/solution/scala/org/apache/flink/training/solutions/ridesandfares/scala/RidesAndFaresSolution.scala
index e5bffe7..f6d9334 100644
--- a/rides-and-fares/src/solution/scala/org/apache/flink/training/solutions/ridesandfares/scala/RidesAndFaresSolution.scala
+++ b/rides-and-fares/src/solution/scala/org/apache/flink/training/solutions/ridesandfares/scala/RidesAndFaresSolution.scala
@@ -36,15 +36,15 @@
* Parameters:
* -rides path-to-input-file
* -fares path-to-input-file
- *
*/
object RidesAndFaresSolution {
+
def main(args: Array[String]) {
// parse parameters
val params = ParameterTool.fromArgs(args)
- val ridesFile = params.get("rides", ExerciseBase.pathToRideData)
- val faresFile = params.get("fares", ExerciseBase.pathToFareData)
+ val ridesFile = params.get("rides", ExerciseBase.PATH_TO_RIDE_DATA)
+ val faresFile = params.get("fares", ExerciseBase.PATH_TO_FARE_DATA)
val delay = 60; // at most 60 seconds of delay
val servingSpeedFactor = 1800 // 30 minutes worth of events are served every second
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 128d4b5..e1d270a 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
@@ -24,15 +24,15 @@
import org.apache.flink.training.exercises.testing.TaxiRideTestBase;
import org.apache.flink.training.solutions.ridesandfares.RidesAndFaresSolution;
-import com.google.common.collect.Lists;
import org.joda.time.DateTime;
import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
import static org.hamcrest.collection.IsIterableContainingInAnyOrder.containsInAnyOrder;
import static org.junit.Assert.assertThat;
-import java.util.ArrayList;
-import java.util.List;
-
public class RidesAndFaresTest extends TaxiRideTestBase<Tuple2<TaxiRide, TaxiFare>> {
static Testable javaExercise = () -> RidesAndFaresExercise.main(new String[]{});
@@ -48,7 +48,7 @@
TestRideSource rides = new TestRideSource(ride1, ride2);
TestFareSource fares = new TestFareSource(fare1, fare2);
- ArrayList<Tuple2<TaxiRide, TaxiFare>> expected = Lists.newArrayList(
+ List<Tuple2<TaxiRide, TaxiFare>> expected = Arrays.asList(
new Tuple2<>(ride1, fare1),
new Tuple2<>(ride2, fare2));
@@ -60,7 +60,7 @@
TestRideSource rides = new TestRideSource(ride1, ride2);
TestFareSource fares = new TestFareSource(fare2, fare1);
- ArrayList<Tuple2<TaxiRide, TaxiFare>> expected = Lists.newArrayList(
+ List<Tuple2<TaxiRide, TaxiFare>> expected = Arrays.asList(
new Tuple2<>(ride1, fare1),
new Tuple2<>(ride2, fare2));
@@ -69,7 +69,7 @@
private TaxiRide testRide(long rideId) {
return new TaxiRide(rideId, true, new DateTime(0), new DateTime(0),
- 0F, 0F, 0F, 0F, (short)1, 0, rideId);
+ 0F, 0F, 0F, 0F, (short) 1, 0, rideId);
}
private TaxiFare testFare(long rideId) {
@@ -81,4 +81,4 @@
return runApp(rides, fares, new TestSink<>(), javaExercise, javaSolution);
}
-}
\ No newline at end of file
+}
