You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gr...@apache.org on 2017/05/31 20:57:16 UTC
[1/5] flink git commit: [FLINK-6777] [shell] Activate strict
checkstyle
Repository: flink
Updated Branches:
refs/heads/master 88545130b -> 49a58fdbb
[FLINK-6777] [shell] Activate strict checkstyle
This closes #4029
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/9d9c53ec
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/9d9c53ec
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/9d9c53ec
Branch: refs/heads/master
Commit: 9d9c53ec142c173693985b52ce31f85bacd9b2d8
Parents: 8854513
Author: Greg Hogan <co...@greghogan.com>
Authored: Tue May 30 15:21:19 2017 -0400
Committer: Greg Hogan <co...@greghogan.com>
Committed: Wed May 31 15:57:52 2017 -0400
----------------------------------------------------------------------
flink-scala-shell/pom.xml | 36 +++
.../org/apache/flink/api/java/JarHelper.java | 228 ++++++++++---------
.../api/java/ScalaShellRemoteEnvironment.java | 7 +-
.../java/ScalaShellRemoteStreamEnvironment.java | 9 +-
.../apache/flink/api/java/FlinkILoopTest.java | 21 +-
5 files changed, 176 insertions(+), 125 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/9d9c53ec/flink-scala-shell/pom.xml
----------------------------------------------------------------------
diff --git a/flink-scala-shell/pom.xml b/flink-scala-shell/pom.xml
index c9acbec..6f490e4 100644
--- a/flink-scala-shell/pom.xml
+++ b/flink-scala-shell/pom.xml
@@ -206,6 +206,42 @@ under the License.
</configuration>
</plugin>
+ <!-- Java Code Style -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <version>2.17</version>
+ <dependencies>
+ <dependency>
+ <groupId>com.puppycrawl.tools</groupId>
+ <artifactId>checkstyle</artifactId>
+ <version>6.19</version>
+ </dependency>
+ </dependencies>
+ <configuration>
+ <configLocation>/tools/maven/strict-checkstyle.xml</configLocation>
+ <suppressionsLocation>/tools/maven/suppressions.xml</suppressionsLocation>
+ <includeTestSourceDirectory>true</includeTestSourceDirectory>
+ <logViolationsToConsole>true</logViolationsToConsole>
+ <failOnViolation>true</failOnViolation>
+ </configuration>
+ <executions>
+ <!--
+ Execute checkstyle after compilation but before tests.
+
+ This ensures that any parsing or type checking errors are from
+ javac, so they look as expected. Beyond that, we want to
+ fail as early as possible.
+ -->
+ <execution>
+ <phase>test-compile</phase>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
</plugins>
</build>
http://git-wip-us.apache.org/repos/asf/flink/blob/9d9c53ec/flink-scala-shell/src/main/java/org/apache/flink/api/java/JarHelper.java
----------------------------------------------------------------------
diff --git a/flink-scala-shell/src/main/java/org/apache/flink/api/java/JarHelper.java b/flink-scala-shell/src/main/java/org/apache/flink/api/java/JarHelper.java
index e00c602..0da9096 100644
--- a/flink-scala-shell/src/main/java/org/apache/flink/api/java/JarHelper.java
+++ b/flink-scala-shell/src/main/java/org/apache/flink/api/java/JarHelper.java
@@ -18,29 +18,25 @@
package org.apache.flink.api.java;
-
import java.io.BufferedOutputStream;
import java.io.File;
+import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
-import java.io.FileInputStream;
import java.io.InputStream;
-
-import java.util.jar.JarOutputStream;
import java.util.jar.JarEntry;
import java.util.jar.JarInputStream;
+import java.util.jar.JarOutputStream;
/**
- * Provides utility services for jarring and unjarring files and directories.
- * Note that a given instance of JarHelper is not threadsafe with respect to
- * multiple jar operations.
+ * Provides utility services for jarring and unjarring files and directories. Note that a given instance of JarHelper is
+ * not threadsafe with respect to multiple jar operations.
*
- * Copied from http://grepcode.com/file_/repo1.maven.org/maven2/org.apache.xmlbeans/xmlbeans/2.4.0/org/apache/xmlbeans/impl/common/JarHelper.java/?v=source
+ * <p>Copied from http://grepcode.com/file_/repo1.maven.org/maven2/org.apache.xmlbeans/xmlbeans/2.4.0/org/apache/xmlbeans/impl/common/JarHelper.java/?v=source
*
- * @author Patrick Calahan <a href="mailto:pcal@bea.com">pcal@bea.com</a>
+ * <p>Author: Patrick Calahan <a href="mailto:pcal@bea.com">pcal@bea.com</a>
*/
-public class JarHelper
-{
+public class JarHelper {
// ========================================================================
// Constants
@@ -50,8 +46,11 @@ public class JarHelper
// Variables
private byte[] mBuffer = new byte[BUFFER_SIZE];
+
private int mByteCount = 0;
+
private boolean mVerbose = false;
+
private String mDestJarName = "";
// ========================================================================
@@ -60,7 +59,8 @@ public class JarHelper
/**
* Instantiates a new JarHelper.
*/
- public JarHelper() {}
+ public JarHelper() {
+ }
// ========================================================================
// Public methods
@@ -71,32 +71,31 @@ public class JarHelper
public void jarDir(File dirOrFile2Jar, File destJar)
throws IOException {
- if (dirOrFile2Jar == null || destJar == null)
- {
- throw new IllegalArgumentException();
- }
+ if (dirOrFile2Jar == null || destJar == null) {
+ throw new IllegalArgumentException();
+ }
- mDestJarName = destJar.getCanonicalPath();
- FileOutputStream fout = new FileOutputStream(destJar);
- JarOutputStream jout = new JarOutputStream(fout);
- //jout.setLevel(0);
- try {
- jarDir(dirOrFile2Jar, jout, null);
- } catch(IOException ioe) {
- throw ioe;
- } finally {
- jout.close();
- fout.close();
- }
+ mDestJarName = destJar.getCanonicalPath();
+ FileOutputStream fout = new FileOutputStream(destJar);
+ JarOutputStream jout = new JarOutputStream(fout);
+ //jout.setLevel(0);
+ try {
+ jarDir(dirOrFile2Jar, jout, null);
+ } catch (IOException ioe) {
+ throw ioe;
+ } finally {
+ jout.close();
+ fout.close();
+ }
}
/**
* Unjars a given jar file into a given directory.
*/
public void unjarDir(File jarFile, File destDir) throws IOException {
- BufferedOutputStream dest = null;
- FileInputStream fis = new FileInputStream(jarFile);
- unjar(fis, destDir);
+ BufferedOutputStream dest = null;
+ FileInputStream fis = new FileInputStream(jarFile);
+ unjar(fis, destDir);
}
/**
@@ -104,111 +103,118 @@ public class JarHelper
* directory.
*/
public void unjar(InputStream in, File destDir) throws IOException {
- BufferedOutputStream dest = null;
- JarInputStream jis = new JarInputStream(in);
- JarEntry entry;
- while ((entry = jis.getNextJarEntry()) != null) {
- if (entry.isDirectory()) {
- File dir = new File(destDir,entry.getName());
- dir.mkdir();
- if (entry.getTime() != -1) {dir.setLastModified(entry.getTime());}
- continue;
- }
- int count;
- byte[] data = new byte[ BUFFER_SIZE ];
- File destFile = new File(destDir, entry.getName());
- if (mVerbose) {
- System.out.println("unjarring " + destFile +
+ BufferedOutputStream dest = null;
+ JarInputStream jis = new JarInputStream(in);
+ JarEntry entry;
+ while ((entry = jis.getNextJarEntry()) != null) {
+ if (entry.isDirectory()) {
+ File dir = new File(destDir, entry.getName());
+ dir.mkdir();
+ if (entry.getTime() != -1) {
+ dir.setLastModified(entry.getTime());
+ }
+ continue;
+ }
+ int count;
+ byte[] data = new byte[BUFFER_SIZE];
+ File destFile = new File(destDir, entry.getName());
+ if (mVerbose) {
+ System.out.println("unjarring " + destFile +
" from " + entry.getName());
- }
- FileOutputStream fos = new FileOutputStream(destFile);
- dest = new BufferedOutputStream(fos, BUFFER_SIZE);
- try {
- while ((count = jis.read(data, 0, BUFFER_SIZE)) != -1) {
- dest.write(data, 0, count);
}
- dest.flush();
- } finally {
- dest.close();
+ FileOutputStream fos = new FileOutputStream(destFile);
+ dest = new BufferedOutputStream(fos, BUFFER_SIZE);
+ try {
+ while ((count = jis.read(data, 0, BUFFER_SIZE)) != -1) {
+ dest.write(data, 0, count);
+ }
+ dest.flush();
+ } finally {
+ dest.close();
+ }
+ if (entry.getTime() != -1) {
+ destFile.setLastModified(entry.getTime());
+ }
}
- if (entry.getTime() != -1) {destFile.setLastModified(entry.getTime());}
- }
- jis.close();
+ jis.close();
}
public void setVerbose(boolean b) {
- mVerbose = b;
+ mVerbose = b;
}
// ========================================================================
// Private methods
private static final char SEP = '/';
+
/**
* Recursively jars up the given path under the given directory.
*/
private void jarDir(File dirOrFile2jar, JarOutputStream jos, String path)
throws IOException {
- if (mVerbose) { System.out.println("checking " + dirOrFile2jar);}
- if (dirOrFile2jar.isDirectory()) {
- String[] dirList = dirOrFile2jar.list();
- String subPath = (path == null) ? "" : (path+dirOrFile2jar.getName()+SEP);
- if (path != null) {
- JarEntry je = new JarEntry(subPath);
- je.setTime(dirOrFile2jar.lastModified());
- jos.putNextEntry(je);
- jos.flush();
- jos.closeEntry();
- }
- for (int i = 0; i < dirList.length; i++) {
- File f = new File(dirOrFile2jar, dirList[i]);
- jarDir(f,jos,subPath);
- }
- } else if (dirOrFile2jar.exists()) {
- if (dirOrFile2jar.getCanonicalPath().equals(mDestJarName))
- {
- if (mVerbose) {System.out.println("skipping " + dirOrFile2jar.getPath());}
- return;
- }
-
if (mVerbose) {
- System.out.println("adding " + dirOrFile2jar.getPath());
- }
- FileInputStream fis = new FileInputStream(dirOrFile2jar);
- try {
- JarEntry entry = new JarEntry(path+dirOrFile2jar.getName());
- entry.setTime(dirOrFile2jar.lastModified());
- jos.putNextEntry(entry);
- while ((mByteCount = fis.read(mBuffer)) != -1) {
- jos.write(mBuffer, 0, mByteCount);
- if (mVerbose) { System.out.println("wrote " + mByteCount + " bytes");}
+ System.out.println("checking " + dirOrFile2jar);
}
- jos.flush();
- jos.closeEntry();
- } catch (IOException ioe) {
- throw ioe;
- } finally {
- fis.close();
+ if (dirOrFile2jar.isDirectory()) {
+ String[] dirList = dirOrFile2jar.list();
+ String subPath = (path == null) ? "" : (path + dirOrFile2jar.getName() + SEP);
+ if (path != null) {
+ JarEntry je = new JarEntry(subPath);
+ je.setTime(dirOrFile2jar.lastModified());
+ jos.putNextEntry(je);
+ jos.flush();
+ jos.closeEntry();
+ }
+ for (int i = 0; i < dirList.length; i++) {
+ File f = new File(dirOrFile2jar, dirList[i]);
+ jarDir(f, jos, subPath);
+ }
+ } else if (dirOrFile2jar.exists()) {
+ if (dirOrFile2jar.getCanonicalPath().equals(mDestJarName)) {
+ if (mVerbose) {
+ System.out.println("skipping " + dirOrFile2jar.getPath());
+ }
+ return;
+ }
+
+ if (mVerbose) {
+ System.out.println("adding " + dirOrFile2jar.getPath());
+ }
+ FileInputStream fis = new FileInputStream(dirOrFile2jar);
+ try {
+ JarEntry entry = new JarEntry(path + dirOrFile2jar.getName());
+ entry.setTime(dirOrFile2jar.lastModified());
+ jos.putNextEntry(entry);
+ while ((mByteCount = fis.read(mBuffer)) != -1) {
+ jos.write(mBuffer, 0, mByteCount);
+ if (mVerbose) {
+ System.out.println("wrote " + mByteCount + " bytes");
+ }
+ }
+ jos.flush();
+ jos.closeEntry();
+ } catch (IOException ioe) {
+ throw ioe;
+ } finally {
+ fis.close();
+ }
}
}
- }
// for debugging
- public static void main(String[] args)
- throws IOException
- {
- if (args.length < 2)
- {
- System.err.println("Usage: JarHelper jarname.jar directory");
- return;
- }
+ public static void main(String[] args) throws IOException {
+ if (args.length < 2) {
+ System.err.println("Usage: JarHelper jarname.jar directory");
+ return;
+ }
- JarHelper jarHelper = new JarHelper();
- jarHelper.mVerbose = true;
+ JarHelper jarHelper = new JarHelper();
+ jarHelper.mVerbose = true;
- File destJar = new File(args[0]);
- File dirOrFile2Jar = new File(args[1]);
+ File destJar = new File(args[0]);
+ File dirOrFile2Jar = new File(args[1]);
- jarHelper.jarDir(dirOrFile2Jar, destJar);
+ jarHelper.jarDir(dirOrFile2Jar, destJar);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/9d9c53ec/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteEnvironment.java b/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteEnvironment.java
index 1465e1d..b8e1c01 100644
--- a/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteEnvironment.java
+++ b/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteEnvironment.java
@@ -20,7 +20,6 @@ package org.apache.flink.api.java;
*/
import org.apache.flink.api.common.PlanExecutor;
-
import org.apache.flink.api.scala.FlinkILoop;
import org.apache.flink.configuration.Configuration;
@@ -40,7 +39,7 @@ public class ScalaShellRemoteEnvironment extends RemoteEnvironment {
private FlinkILoop flinkILoop;
/**
- * Creates new ScalaShellRemoteEnvironment that has a reference to the FlinkILoop
+ * Creates new ScalaShellRemoteEnvironment that has a reference to the FlinkILoop.
*
* @param host The host name or address of the master (JobManager), where the program should be executed.
* @param port The port of the master (JobManager), where the program should be executed.
@@ -83,7 +82,7 @@ public class ScalaShellRemoteEnvironment extends RemoteEnvironment {
}
public static void disableAllContextAndOtherEnvironments() {
-
+
// we create a context environment that prevents the instantiation of further
// context environments. at the same time, setting the context environment prevents manual
// creation of local and remote environments
@@ -96,7 +95,7 @@ public class ScalaShellRemoteEnvironment extends RemoteEnvironment {
};
initializeContextEnvironment(factory);
}
-
+
public static void resetContextEnvironments() {
ExecutionEnvironment.resetContextEnvironment();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/9d9c53ec/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteStreamEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteStreamEnvironment.java b/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteStreamEnvironment.java
index 43c8174..22b280f 100644
--- a/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteStreamEnvironment.java
+++ b/flink-scala-shell/src/main/java/org/apache/flink/api/java/ScalaShellRemoteStreamEnvironment.java
@@ -15,6 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.flink.api.java;
import org.apache.flink.api.common.JobExecutionResult;
@@ -25,6 +26,7 @@ import org.apache.flink.streaming.api.environment.RemoteStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironmentFactory;
import org.apache.flink.streaming.api.graph.StreamGraph;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,12 +35,15 @@ import java.net.URL;
import java.util.ArrayList;
import java.util.List;
-
+/**
+ * A {@link RemoteStreamEnvironment} for the Scala shell.
+ */
public class ScalaShellRemoteStreamEnvironment extends RemoteStreamEnvironment {
private static final Logger LOG = LoggerFactory.getLogger(ScalaShellRemoteStreamEnvironment.class);
// reference to Scala Shell, for access to virtual directory
private FlinkILoop flinkILoop;
+
/**
* Creates a new RemoteStreamEnvironment that points to the master
* (JobManager) described by the given host name and port.
@@ -62,6 +67,7 @@ public class ScalaShellRemoteStreamEnvironment extends RemoteStreamEnvironment {
super(host, port, configuration, jarFiles);
this.flinkILoop = flinkILoop;
}
+
/**
* Executes the remote job.
*
@@ -98,7 +104,6 @@ public class ScalaShellRemoteStreamEnvironment extends RemoteStreamEnvironment {
initializeContextEnvironment(factory);
}
-
public static void disableAllContextAndOtherEnvironments() {
// we create a context environment that prevents the instantiation of further
// context environments. at the same time, setting the context environment prevents manual
http://git-wip-us.apache.org/repos/asf/flink/blob/9d9c53ec/flink-scala-shell/src/test/java/org/apache/flink/api/java/FlinkILoopTest.java
----------------------------------------------------------------------
diff --git a/flink-scala-shell/src/test/java/org/apache/flink/api/java/FlinkILoopTest.java b/flink-scala-shell/src/test/java/org/apache/flink/api/java/FlinkILoopTest.java
index 919f805..6893814 100644
--- a/flink-scala-shell/src/test/java/org/apache/flink/api/java/FlinkILoopTest.java
+++ b/flink-scala-shell/src/test/java/org/apache/flink/api/java/FlinkILoopTest.java
@@ -28,6 +28,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.RemoteStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.TestLogger;
+
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.BDDMockito;
@@ -37,15 +38,19 @@ import org.mockito.stubbing.Answer;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.List;
+
import scala.Option;
import scala.tools.nsc.Settings;
import scala.tools.nsc.settings.MutableSettings;
-import java.util.List;
-
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
+/**
+ * Integration tests for {@link FlinkILoop}.
+ */
@RunWith(PowerMockRunner.class)
@PrepareForTest(PlanExecutor.class)
public class FlinkILoopTest extends TestLogger {
@@ -68,18 +73,18 @@ public class FlinkILoopTest extends TestLogger {
)).willAnswer(new Answer<PlanExecutor>() {
@Override
public PlanExecutor answer(InvocationOnMock invocation) throws Throwable {
- testPlanExecutor.setHost((String)invocation.getArguments()[0]);
- testPlanExecutor.setPort((Integer)invocation.getArguments()[1]);
- testPlanExecutor.setConfiguration((Configuration)invocation.getArguments()[2]);
- testPlanExecutor.setJars((List<String>)invocation.getArguments()[3]);
- testPlanExecutor.setGlobalClasspaths((List<String>)invocation.getArguments()[4]);
+ testPlanExecutor.setHost((String) invocation.getArguments()[0]);
+ testPlanExecutor.setPort((Integer) invocation.getArguments()[1]);
+ testPlanExecutor.setConfiguration((Configuration) invocation.getArguments()[2]);
+ testPlanExecutor.setJars((List<String>) invocation.getArguments()[3]);
+ testPlanExecutor.setGlobalClasspaths((List<String>) invocation.getArguments()[4]);
return testPlanExecutor;
}
});
Settings settings = new Settings();
- ((MutableSettings.BooleanSetting)settings.usejavacp()).value_$eq(true);
+ ((MutableSettings.BooleanSetting) settings.usejavacp()).value_$eq(true);
flinkILoop.settings_$eq(settings);
flinkILoop.createInterpreter();
[2/5] flink git commit: [FLINK-6779] [scala] Activate strict
checkstyle
Posted by gr...@apache.org.
[FLINK-6779] [scala] Activate strict checkstyle
This closes #4030
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6ab7719b
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6ab7719b
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6ab7719b
Branch: refs/heads/master
Commit: 6ab7719b565934a0c127dcca3f46d71a71ecc0ef
Parents: 9d9c53e
Author: Greg Hogan <co...@greghogan.com>
Authored: Tue May 30 15:35:33 2017 -0400
Committer: Greg Hogan <co...@greghogan.com>
Committed: Wed May 31 15:58:47 2017 -0400
----------------------------------------------------------------------
flink-scala/pom.xml | 36 ++++++++++++++++++++
.../scala/operators/ScalaAggregateOperator.java | 27 ++++++---------
.../scala/operators/ScalaCsvOutputFormat.java | 15 ++++----
3 files changed, 53 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6ab7719b/flink-scala/pom.xml
----------------------------------------------------------------------
diff --git a/flink-scala/pom.xml b/flink-scala/pom.xml
index 7fad487..1c4d7c4 100644
--- a/flink-scala/pom.xml
+++ b/flink-scala/pom.xml
@@ -233,6 +233,42 @@ under the License.
</executions>
</plugin>
+ <!-- Java check style -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <version>2.17</version>
+ <dependencies>
+ <dependency>
+ <groupId>com.puppycrawl.tools</groupId>
+ <artifactId>checkstyle</artifactId>
+ <version>6.19</version>
+ </dependency>
+ </dependencies>
+ <configuration>
+ <configLocation>/tools/maven/strict-checkstyle.xml</configLocation>
+ <suppressionsLocation>/tools/maven/suppressions.xml</suppressionsLocation>
+ <includeTestSourceDirectory>true</includeTestSourceDirectory>
+ <logViolationsToConsole>true</logViolationsToConsole>
+ <failOnViolation>true</failOnViolation>
+ </configuration>
+ <executions>
+ <!--
+ Execute checkstyle after compilation but before tests.
+
+ This ensures that any parsing or type checking errors are from
+ javac, so they look as expected. Beyond that, we want to
+ fail as early as possible.
+ -->
+ <execution>
+ <phase>test-compile</phase>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
</plugins>
</build>
http://git-wip-us.apache.org/repos/asf/flink/blob/6ab7719b/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java b/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java
index 80e91ac..e433dcc 100644
--- a/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java
+++ b/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java
@@ -18,15 +18,13 @@
package org.apache.flink.api.scala.operators;
-import java.util.ArrayList;
-import java.util.List;
-
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.Public;
import org.apache.flink.api.common.InvalidProgramException;
import org.apache.flink.api.common.functions.GroupCombineFunction;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.common.functions.RichGroupReduceFunction;
+import org.apache.flink.api.common.operators.Keys;
import org.apache.flink.api.common.operators.Operator;
import org.apache.flink.api.common.operators.SingleInputSemanticProperties;
import org.apache.flink.api.common.operators.UnaryOperatorInformation;
@@ -36,7 +34,6 @@ import org.apache.flink.api.java.aggregation.AggregationFunction;
import org.apache.flink.api.java.aggregation.AggregationFunctionFactory;
import org.apache.flink.api.java.aggregation.Aggregations;
import org.apache.flink.api.java.operators.Grouping;
-import org.apache.flink.api.common.operators.Keys;
import org.apache.flink.api.java.operators.SingleInputOperator;
import org.apache.flink.api.java.typeutils.TupleTypeInfoBase;
import org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase;
@@ -44,6 +41,9 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.apache.flink.util.Preconditions;
+import java.util.ArrayList;
+import java.util.List;
+
import scala.Product;
/**
@@ -62,8 +62,7 @@ public class ScalaAggregateOperator<IN> extends SingleInputOperator<IN, IN, Scal
private final Grouping<IN> grouping;
/**
- * <p>
- * Non grouped aggregation
+ * Non grouped aggregation.
*/
public ScalaAggregateOperator(org.apache.flink.api.java.DataSet<IN> input, Aggregations function, int field) {
super(Preconditions.checkNotNull(input), input.getType());
@@ -90,8 +89,7 @@ public class ScalaAggregateOperator<IN> extends SingleInputOperator<IN, IN, Scal
}
/**
- *
- * Grouped aggregation
+ * Grouped aggregation.
*
* @param input
* @param function
@@ -121,7 +119,6 @@ public class ScalaAggregateOperator<IN> extends SingleInputOperator<IN, IN, Scal
this.grouping = input;
}
-
public ScalaAggregateOperator<IN> and(Aggregations function, int field) {
Preconditions.checkNotNull(function);
@@ -131,7 +128,6 @@ public class ScalaAggregateOperator<IN> extends SingleInputOperator<IN, IN, Scal
throw new IllegalArgumentException("Aggregation field position is out of range.");
}
-
AggregationFunctionFactory factory = function.getFactory();
AggregationFunction<?> aggFunct = factory.createAggregationFunction(inType.getTypeAt(field).getTypeClass());
@@ -150,7 +146,6 @@ public class ScalaAggregateOperator<IN> extends SingleInputOperator<IN, IN, Scal
throw new IllegalStateException();
}
-
// construct the aggregation function
AggregationFunction<Object>[] aggFunctions = new AggregationFunction[this.aggregationFunctions.size()];
int[] fields = new int[this.fields.size()];
@@ -162,12 +157,11 @@ public class ScalaAggregateOperator<IN> extends SingleInputOperator<IN, IN, Scal
genName.append(aggFunctions[i].toString()).append('(').append(fields[i]).append(')').append(',');
}
- genName.setLength(genName.length()-1);
+ genName.setLength(genName.length() - 1);
@SuppressWarnings("rawtypes")
RichGroupReduceFunction<IN, IN> function = new AggregatingUdf(getInputType(), aggFunctions, fields);
-
String name = getName() != null ? getName() : genName.toString();
// distinguish between grouped reduce and non-grouped reduce
@@ -234,10 +228,10 @@ public class ScalaAggregateOperator<IN> extends SingleInputOperator<IN, IN, Scal
// --------------------------------------------------------------------------------------------
@Internal
- public static final class AggregatingUdf<T extends Product>
+ private static final class AggregatingUdf<T extends Product>
extends RichGroupReduceFunction<T, T>
- implements GroupCombineFunction<T, T>
- {
+ implements GroupCombineFunction<T, T> {
+
private static final long serialVersionUID = 1L;
private final int[] fieldPositions;
@@ -258,7 +252,6 @@ public class ScalaAggregateOperator<IN> extends SingleInputOperator<IN, IN, Scal
this.fieldPositions = fieldPositions;
}
-
@Override
public void open(Configuration parameters) throws Exception {
for (AggregationFunction<Object> aggFunction : aggFunctions) {
http://git-wip-us.apache.org/repos/asf/flink/blob/6ab7719b/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvOutputFormat.java
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvOutputFormat.java b/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvOutputFormat.java
index abc3ffc..ee5ba65 100644
--- a/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvOutputFormat.java
+++ b/flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaCsvOutputFormat.java
@@ -16,7 +16,6 @@
* limitations under the License.
*/
-
package org.apache.flink.api.scala.operators;
import org.apache.flink.annotation.PublicEvolving;
@@ -29,13 +28,14 @@ import org.apache.flink.api.java.typeutils.InputTypeConfigurable;
import org.apache.flink.core.fs.Path;
import org.apache.flink.types.StringValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.Writer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import scala.Product;
/**
@@ -124,8 +124,8 @@ public class ScalaCsvOutputFormat<T extends Product> extends FileOutputFormat<T>
/**
* Configures the format to either allow null values (writing an empty field),
* or to throw an exception when encountering a null field.
- * <p>
- * by default, null values are allowed.
+ *
+ * <p>By default, null values are allowed.
*
* @param allowNulls Flag to indicate whether the output format should accept null values.
*/
@@ -147,8 +147,8 @@ public class ScalaCsvOutputFormat<T extends Product> extends FileOutputFormat<T>
* Configures whether the output format should quote string values. String values are fields
* of type {@link String} and {@link org.apache.flink.types.StringValue}, as well as
* all subclasses of the latter.
- * <p>
- * By default, strings are not quoted.
+ *
+ * <p>By default, strings are not quoted.
*
* @param quoteStrings Flag indicating whether string fields should be quoted.
*/
@@ -217,7 +217,6 @@ public class ScalaCsvOutputFormat<T extends Product> extends FileOutputFormat<T>
}
/**
- *
* The purpose of this method is solely to check whether the data type to be processed
* is in fact a tuple type.
*/
[5/5] flink git commit: [FLINK-6332] [build] Upgrade Scala versions
Posted by gr...@apache.org.
[FLINK-6332] [build] Upgrade Scala versions
Upgrade to the last maintenance releases of Scala 2.10 and 2.11.
This closes #3957
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/49a58fdb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/49a58fdb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/49a58fdb
Branch: refs/heads/master
Commit: 49a58fdbb4ac80a0c0d34e42803ee6d1ca26b77a
Parents: 52efdcb
Author: Greg Hogan <co...@greghogan.com>
Authored: Fri May 19 13:05:55 2017 -0400
Committer: Greg Hogan <co...@greghogan.com>
Committed: Wed May 31 16:56:29 2017 -0400
----------------------------------------------------------------------
pom.xml | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/49a58fdb/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 1b6307e..ae92be9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -97,9 +97,9 @@ under the License.
<guava.version>18.0</guava.version>
<akka.version>2.3-custom</akka.version>
<java.version>1.7</java.version>
- <scala.macros.version>2.0.1</scala.macros.version>
+ <scala.macros.version>2.1.0</scala.macros.version>
<!-- Default scala versions, may be overwritten by build profiles -->
- <scala.version>2.10.4</scala.version>
+ <scala.version>2.10.6</scala.version>
<scala.binary.version>2.10</scala.binary.version>
<chill.version>0.7.4</chill.version>
<asm.version>5.0.4</asm.version>
@@ -488,7 +488,7 @@ under the License.
</property>
</activation>
<properties>
- <scala.version>2.11.7</scala.version>
+ <scala.version>2.11.11</scala.version>
<scala.binary.version>2.11</scala.binary.version>
</properties>
</profile>
[3/5] flink git commit: [FLINK-6778] [dist] Activate strict checkstyle
Posted by gr...@apache.org.
[FLINK-6778] [dist] Activate strict checkstyle
This closes #4031
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b5eac06d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b5eac06d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b5eac06d
Branch: refs/heads/master
Commit: b5eac06d7cca8317a0f10e08bd5aae654a587a13
Parents: 6ab7719
Author: Greg Hogan <co...@greghogan.com>
Authored: Tue May 30 15:26:19 2017 -0400
Committer: Greg Hogan <co...@greghogan.com>
Committed: Wed May 31 15:59:09 2017 -0400
----------------------------------------------------------------------
flink-dist/pom.xml | 36 ++++++++++++++++++++
...kManagerHeapSizeCalculationJavaBashTest.java | 5 +--
2 files changed, 39 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b5eac06d/flink-dist/pom.xml
----------------------------------------------------------------------
diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
index b037e55..c7af451 100644
--- a/flink-dist/pom.xml
+++ b/flink-dist/pom.xml
@@ -530,6 +530,42 @@ under the License.
</configuration>
</plugin>
+ <!-- Java Code Style -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <version>2.17</version>
+ <dependencies>
+ <dependency>
+ <groupId>com.puppycrawl.tools</groupId>
+ <artifactId>checkstyle</artifactId>
+ <version>6.19</version>
+ </dependency>
+ </dependencies>
+ <configuration>
+ <configLocation>/tools/maven/strict-checkstyle.xml</configLocation>
+ <suppressionsLocation>/tools/maven/suppressions.xml</suppressionsLocation>
+ <includeTestSourceDirectory>true</includeTestSourceDirectory>
+ <logViolationsToConsole>true</logViolationsToConsole>
+ <failOnViolation>true</failOnViolation>
+ </configuration>
+ <executions>
+ <!--
+ Execute checkstyle after compilation but before tests.
+
+ This ensures that any parsing or type checking errors are from
+ javac, so they look as expected. Beyond that, we want to
+ fail as early as possible.
+ -->
+ <execution>
+ <phase>test-compile</phase>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
</plugins>
</build>
http://git-wip-us.apache.org/repos/asf/flink/blob/b5eac06d/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java
----------------------------------------------------------------------
diff --git a/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java b/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java
index 11d8ec7..c037426 100644
--- a/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java
+++ b/flink-dist/src/test/java/org/apache/flink/dist/TaskManagerHeapSizeCalculationJavaBashTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.taskexecutor.TaskManagerServices;
import org.apache.flink.util.OperatingSystem;
import org.apache.flink.util.TestLogger;
+
import org.junit.Assume;
import org.junit.Before;
import org.junit.Test;
@@ -43,7 +44,7 @@ import static org.junit.Assert.assertThat;
* <tt>taskmanager.sh</tt> returns the same values as the heap size calculation of
* {@link TaskManagerServices#calculateHeapSizeMB(long, Configuration)}.
*
- * NOTE: the shell script uses <tt>awk</tt> to perform floating-point arithmetic which uses
+ * <p>NOTE: the shell script uses <tt>awk</tt> to perform floating-point arithmetic which uses
* <tt>double</tt> precision but our Java code restrains to <tt>float</tt> because we actually do
* not need high precision.
*/
@@ -55,7 +56,7 @@ public class TaskManagerHeapSizeCalculationJavaBashTest extends TestLogger {
/**
* Number of tests with random values.
*
- * NOTE: calling the external test script is slow and thus low numbers are preferred for general
+ * <p>NOTE: calling the external test script is slow and thus low numbers are preferred for general
* testing.
*/
private static final int NUM_RANDOM_TESTS = 20;
[4/5] flink git commit: [FLINK-6560] [build] Restore maven
parallelism in flink-tests
Posted by gr...@apache.org.
[FLINK-6560] [build] Restore maven parallelism in flink-tests
Configure a default value for Maven variable 'flink.forkCountTestPackage'.
This closes #3875
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/52efdcb7
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/52efdcb7
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/52efdcb7
Branch: refs/heads/master
Commit: 52efdcb7dc2cfa8a79852f06178db5824eb83fb8
Parents: b5eac06
Author: Greg Hogan <co...@greghogan.com>
Authored: Thu May 11 13:44:20 2017 -0400
Committer: Greg Hogan <co...@greghogan.com>
Committed: Wed May 31 16:08:16 2017 -0400
----------------------------------------------------------------------
pom.xml | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/52efdcb7/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 95fcf1e..1b6307e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -89,6 +89,9 @@ under the License.
forkCount is not exposed as a property. With this we can set
it on the "mvn" commandline in travis. -->
<flink.forkCount>1C</flink.forkCount>
+ <!-- Allow overriding the fork behaviour for the expensive tests in flink-tests
+ to avoid process kills due to container limits on TravisCI -->
+ <flink.forkCountTestPackage>${flink.forkCount}</flink.forkCountTestPackage>
<flink.reuseForks>true</flink.reuseForks>
<log4j.configuration>log4j-test.properties</log4j.configuration>
<guava.version>18.0</guava.version>