You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by uc...@apache.org on 2015/08/26 15:47:20 UTC
flink git commit: [FLINK-2555] Properly pass security credentials in
the Hadoop Input/Output format wrappers
Repository: flink
Updated Branches:
refs/heads/master 6e1de9889 -> b264b0100
[FLINK-2555] Properly pass security credentials in the Hadoop Input/Output format wrappers
This is needed because the Hadoop IF/OF's are using Hadoop's FileSystem stack, which is using
the security credentials passed in the JobConf / Job class in the getSplits() method.
Note that access to secured Hadoop 1.x using Hadoop IF/OF's is not possible with this change.
This limitation is due to missing methods in the old APIs.
- Add some comments & change dependency scope to test
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/b264b010
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/b264b010
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/b264b010
Branch: refs/heads/master
Commit: b264b010062575428f7301a31495df0171c9505e
Parents: 6e1de98
Author: Robert Metzger <rm...@apache.org>
Authored: Thu Aug 20 18:43:04 2015 +0200
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Wed Aug 26 15:22:01 2015 +0200
----------------------------------------------------------------------
flink-core/pom.xml | 6 ++
flink-dist/pom.xml | 1 -
flink-java/pom.xml | 3 +-
.../flink/api/java/ExecutionEnvironment.java | 12 +--
.../common/HadoopInputFormatCommonBase.java | 77 ++++++++++++++++++++
.../common/HadoopOutputFormatCommonBase.java | 47 ++++++++++++
.../java/hadoop/mapred/HadoopInputFormat.java | 8 ++
.../hadoop/mapred/HadoopInputFormatBase.java | 26 ++++++-
.../java/hadoop/mapred/HadoopOutputFormat.java | 8 ++
.../hadoop/mapred/HadoopOutputFormatBase.java | 28 +++++--
.../hadoop/mapreduce/HadoopInputFormat.java | 6 ++
.../hadoop/mapreduce/HadoopInputFormatBase.java | 63 ++++++++++------
.../hadoop/mapreduce/HadoopOutputFormat.java | 6 ++
.../mapreduce/HadoopOutputFormatBase.java | 28 +++++--
.../flink/runtime/fs/hdfs/HadoopFileSystem.java | 2 +-
.../flink/api/scala/ExecutionEnvironment.scala | 12 +--
flink-shaded-hadoop/pom.xml | 1 -
.../flink-connector-flume/pom.xml | 1 -
.../flink-connector-twitter/pom.xml | 1 -
flink-yarn-tests/pom.xml | 1 -
pom.xml | 7 +-
21 files changed, 285 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-core/pom.xml
----------------------------------------------------------------------
diff --git a/flink-core/pom.xml b/flink-core/pom.xml
index d59e755..7ff0fbc 100644
--- a/flink-core/pom.xml
+++ b/flink-core/pom.xml
@@ -64,6 +64,12 @@ under the License.
<version>${guava.version}</version>
</dependency>
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
<build>
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-dist/pom.xml
----------------------------------------------------------------------
diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
index 2264119..d0548b7 100644
--- a/flink-dist/pom.xml
+++ b/flink-dist/pom.xml
@@ -305,7 +305,6 @@ under the License.
<!--Build uber jar-->
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
- <version>2.3</version>
<executions>
<execution>
<phase>package</phase>
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/pom.xml
----------------------------------------------------------------------
diff --git a/flink-java/pom.xml b/flink-java/pom.xml
index d777048..6b818e0 100644
--- a/flink-java/pom.xml
+++ b/flink-java/pom.xml
@@ -40,7 +40,6 @@ under the License.
<artifactId>flink-core</artifactId>
<version>${project.version}</version>
</dependency>
-
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>${shading-artifact.name}</artifactId>
@@ -79,7 +78,7 @@ under the License.
<dependency>
<groupId>de.javakaffee</groupId>
<artifactId>kryo-serializers</artifactId>
- <version>0.27</version>
+ <version>0.36</version>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
index d50ddb4..d3d8192 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/ExecutionEnvironment.java
@@ -569,8 +569,8 @@ public abstract class ExecutionEnvironment {
* Creates a {@link DataSet} from the given {@link org.apache.hadoop.mapreduce.lib.input.FileInputFormat}. The
* given inputName is set on the given job.
*/
- public <K,V> DataSource<Tuple2<K, V>> readHadoopFile(org.apache.hadoop.mapreduce.lib.input.FileInputFormat<K,V> mapredInputFormat, Class<K> key, Class<V> value, String inputPath, Job job) throws IOException {
- DataSource<Tuple2<K, V>> result = createHadoopInput(mapredInputFormat, key, value, job);
+ public <K,V> DataSource<Tuple2<K, V>> readHadoopFile(org.apache.hadoop.mapreduce.lib.input.FileInputFormat<K,V> mapreduceInputFormat, Class<K> key, Class<V> value, String inputPath, Job job) throws IOException {
+ DataSource<Tuple2<K, V>> result = createHadoopInput(mapreduceInputFormat, key, value, job);
org.apache.hadoop.mapreduce.lib.input.FileInputFormat.addInputPath(job, new org.apache
.hadoop.fs.Path(inputPath));
@@ -582,15 +582,15 @@ public abstract class ExecutionEnvironment {
* Creates a {@link DataSet} from the given {@link org.apache.hadoop.mapreduce.lib.input.FileInputFormat}. A
* {@link org.apache.hadoop.mapreduce.Job} with the given inputPath is created.
*/
- public <K,V> DataSource<Tuple2<K, V>> readHadoopFile(org.apache.hadoop.mapreduce.lib.input.FileInputFormat<K,V> mapredInputFormat, Class<K> key, Class<V> value, String inputPath) throws IOException {
- return readHadoopFile(mapredInputFormat, key, value, inputPath, Job.getInstance());
+ public <K,V> DataSource<Tuple2<K, V>> readHadoopFile(org.apache.hadoop.mapreduce.lib.input.FileInputFormat<K,V> mapreduceInputFormat, Class<K> key, Class<V> value, String inputPath) throws IOException {
+ return readHadoopFile(mapreduceInputFormat, key, value, inputPath, Job.getInstance());
}
/**
* Creates a {@link DataSet} from the given {@link org.apache.hadoop.mapreduce.InputFormat}.
*/
- public <K,V> DataSource<Tuple2<K, V>> createHadoopInput(org.apache.hadoop.mapreduce.InputFormat<K,V> mapredInputFormat, Class<K> key, Class<V> value, Job job) {
- org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat<K, V> hadoopInputFormat = new org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat<K, V>(mapredInputFormat, key, value, job);
+ public <K,V> DataSource<Tuple2<K, V>> createHadoopInput(org.apache.hadoop.mapreduce.InputFormat<K,V> mapreduceInputFormat, Class<K> key, Class<V> value, Job job) {
+ org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat<K, V> hadoopInputFormat = new org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormat<K, V>(mapreduceInputFormat, key, value, job);
return this.createInput(hadoopInputFormat);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopInputFormatCommonBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopInputFormatCommonBase.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopInputFormatCommonBase.java
new file mode 100644
index 0000000..e442943
--- /dev/null
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopInputFormatCommonBase.java
@@ -0,0 +1,77 @@
+/*
+ * 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.
+ */
+package org.apache.flink.api.java.hadoop.common;
+
+import org.apache.flink.api.common.io.RichInputFormat;
+import org.apache.flink.core.io.InputSplit;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+
+/**
+ * A common base for both "mapred" and "mapreduce" Hadoop input formats.
+ */
+public abstract class HadoopInputFormatCommonBase<T, SPITTYPE extends InputSplit> extends RichInputFormat<T, SPITTYPE> {
+ protected transient Credentials credentials;
+
+ protected HadoopInputFormatCommonBase(Credentials creds) {
+ this.credentials = creds;
+ }
+
+ protected void write(ObjectOutputStream out) throws IOException {
+ this.credentials.write(out);
+ }
+
+ public void read(ObjectInputStream in) throws IOException {
+ this.credentials = new Credentials();
+ credentials.readFields(in);
+ }
+
+ /**
+ * This method only exists because there is no UserGroupInformation.getCredentials() method
+ * in Hadoop 1.x
+ *
+ * Note that this method returns "null" in Hadoop 1.x environments.
+ *
+ * @param ugi The user information
+ * @return new credentials object from the user information. MAY RETURN NULL!
+ */
+ public static Credentials getCredentialsFromUGI(UserGroupInformation ugi) {
+ Method getCredentialsMethod = null;
+ for(Method m: ugi.getClass().getMethods()) {
+ if(m.getName().equals("getCredentials")) {
+ getCredentialsMethod = m;
+ break;
+ }
+ }
+ if(getCredentialsMethod == null) {
+ return null;
+ } else {
+ try {
+ return (Credentials) getCredentialsMethod.invoke(ugi);
+ } catch (InvocationTargetException | IllegalAccessException e) {
+ throw new RuntimeException("Unable to get credentials from UserGroupInformation. This is only supported by Hadoop 2.2.0+");
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopOutputFormatCommonBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopOutputFormatCommonBase.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopOutputFormatCommonBase.java
new file mode 100644
index 0000000..de611ce
--- /dev/null
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/common/HadoopOutputFormatCommonBase.java
@@ -0,0 +1,47 @@
+/*
+ * 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.
+ */
+package org.apache.flink.api.java.hadoop.common;
+
+import org.apache.flink.api.common.io.RichOutputFormat;
+import org.apache.hadoop.security.Credentials;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+
+/**
+ * A common base for both "mapred" and "mapreduce" Hadoop output formats.
+ *
+ * The base is taking care of handling (serializing) security credentials.
+ */
+public abstract class HadoopOutputFormatCommonBase<T> extends RichOutputFormat<T> {
+ protected transient Credentials credentials;
+
+ protected HadoopOutputFormatCommonBase(Credentials creds) {
+ this.credentials = creds;
+ }
+
+ protected void write(ObjectOutputStream out) throws IOException {
+ this.credentials.write(out);
+ }
+
+ public void read(ObjectInputStream in) throws IOException {
+ this.credentials = new Credentials();
+ credentials.readFields(in);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormat.java
index 64cf9da..126b629 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormat.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormat.java
@@ -27,6 +27,14 @@ import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.hadoop.mapred.JobConf;
+/**
+ * Wrapper for using HadoopInputFormats (mapred-variant) with Flink.
+ *
+ * The IF is returning a Tuple2<K,V>.
+ *
+ * @param <K> Type of the key
+ * @param <V> Type of the value.
+ */
public class HadoopInputFormat<K, V> extends HadoopInputFormatBase<K, V, Tuple2<K,V>> implements ResultTypeQueryable<Tuple2<K,V>> {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatBase.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatBase.java
index b24e351..932b7de 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatBase.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopInputFormatBase.java
@@ -16,13 +16,12 @@
* limitations under the License.
*/
-
package org.apache.flink.api.java.hadoop.mapred;
import org.apache.flink.api.common.io.FileInputFormat.FileBaseStatistics;
-import org.apache.flink.api.common.io.RichInputFormat;
import org.apache.flink.api.common.io.LocatableInputSplitAssigner;
import org.apache.flink.api.common.io.statistics.BaseStatistics;
+import org.apache.flink.api.java.hadoop.common.HadoopInputFormatCommonBase;
import org.apache.flink.api.java.hadoop.mapred.utils.HadoopUtils;
import org.apache.flink.api.java.hadoop.mapred.wrapper.HadoopDummyReporter;
import org.apache.flink.api.java.hadoop.mapred.wrapper.HadoopInputSplit;
@@ -36,6 +35,8 @@ import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobConfigurable;
import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.ReflectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,7 +46,14 @@ import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
-public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T, HadoopInputSplit> {
+/**
+ * Common base for Java and Scala API for using Hadoop input formats with Flink.
+ *
+ * @param <K> Type of key
+ * @param <V> Type of value
+ * @param <T> The type iself
+ */
+public abstract class HadoopInputFormatBase<K, V, T> extends HadoopInputFormatCommonBase<T, HadoopInputSplit> {
private static final long serialVersionUID = 1L;
@@ -64,7 +72,7 @@ public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T,
protected transient boolean hasNext;
public HadoopInputFormatBase(org.apache.hadoop.mapred.InputFormat<K, V> mapredInputFormat, Class<K> key, Class<V> value, JobConf job) {
- super();
+ super(job.getCredentials());
this.mapredInputFormat = mapredInputFormat;
this.keyClass = key;
this.valueClass = value;
@@ -225,6 +233,7 @@ public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T,
// --------------------------------------------------------------------------------------------
private void writeObject(ObjectOutputStream out) throws IOException {
+ super.write(out);
out.writeUTF(mapredInputFormat.getClass().getName());
out.writeUTF(keyClass.getName());
out.writeUTF(valueClass.getName());
@@ -233,6 +242,8 @@ public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T,
@SuppressWarnings("unchecked")
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ super.read(in);
+
String hadoopInputFormatClassName = in.readUTF();
String keyClassName = in.readUTF();
String valueClassName = in.readUTF();
@@ -256,5 +267,12 @@ public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T,
throw new RuntimeException("Unable to find value class.", e);
}
ReflectionUtils.setConf(mapredInputFormat, jobConf);
+
+ jobConf.getCredentials().addAll(this.credentials);
+ Credentials currentUserCreds = getCredentialsFromUGI(UserGroupInformation.getCurrentUser());
+ if(currentUserCreds != null) {
+ jobConf.getCredentials().addAll(currentUserCreds);
+ }
}
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormat.java
index 0b7368c..9c4c6d7 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormat.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormat.java
@@ -24,6 +24,14 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCommitter;
+/**
+ * Wrapper for using HadoopOutputFormats (mapred-variant) with Flink.
+ *
+ * The IF is returning a Tuple2<K,V>.
+ *
+ * @param <K> Type of the key
+ * @param <V> Type of the value.
+ */
public class HadoopOutputFormat<K,V> extends HadoopOutputFormatBase<K, V, Tuple2<K, V>> {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatBase.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatBase.java
index a1bb906..456003f 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatBase.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapred/HadoopOutputFormatBase.java
@@ -16,11 +16,10 @@
* limitations under the License.
*/
-
package org.apache.flink.api.java.hadoop.mapred;
import org.apache.flink.api.common.io.FinalizeOnMaster;
-import org.apache.flink.api.common.io.RichOutputFormat;
+import org.apache.flink.api.java.hadoop.common.HadoopOutputFormatCommonBase;
import org.apache.flink.api.java.hadoop.mapred.utils.HadoopUtils;
import org.apache.flink.api.java.hadoop.mapred.wrapper.HadoopDummyProgressable;
import org.apache.flink.api.java.hadoop.mapred.wrapper.HadoopDummyReporter;
@@ -34,14 +33,24 @@ import org.apache.hadoop.mapred.OutputCommitter;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.TaskAttemptContext;
import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.ReflectionUtils;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import static org.apache.flink.api.java.hadoop.common.HadoopInputFormatCommonBase.getCredentialsFromUGI;
-public abstract class HadoopOutputFormatBase<K, V, T> extends RichOutputFormat<T> implements FinalizeOnMaster {
+/**
+ * Common base for the mapred HadoopOutputFormat wrappers. There are implementations for Java and Scala.
+ *
+ * @param <K> Type of Key
+ * @param <V> Type of Value
+ * @param <T> Record type.
+ */
+public abstract class HadoopOutputFormatBase<K, V, T> extends HadoopOutputFormatCommonBase<T> implements FinalizeOnMaster {
private static final long serialVersionUID = 1L;
@@ -50,9 +59,9 @@ public abstract class HadoopOutputFormatBase<K, V, T> extends RichOutputFormat<T
protected transient RecordWriter<K,V> recordWriter;
private transient OutputCommitter outputCommitter;
private transient TaskAttemptContext context;
- private transient JobContext jobContext;
public HadoopOutputFormatBase(org.apache.hadoop.mapred.OutputFormat<K, V> mapredOutputFormat, JobConf job) {
+ super(job.getCredentials());
this.mapredOutputFormat = mapredOutputFormat;
HadoopUtils.mergeHadoopConf(job);
this.jobConf = job;
@@ -108,8 +117,9 @@ public abstract class HadoopOutputFormatBase<K, V, T> extends RichOutputFormat<T
this.outputCommitter = this.jobConf.getOutputCommitter();
+ JobContext jobContext;
try {
- this.jobContext = HadoopUtils.instantiateJobContext(this.jobConf, new JobID());
+ jobContext = HadoopUtils.instantiateJobContext(this.jobConf, new JobID());
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -151,12 +161,14 @@ public abstract class HadoopOutputFormatBase<K, V, T> extends RichOutputFormat<T
// --------------------------------------------------------------------------------------------
private void writeObject(ObjectOutputStream out) throws IOException {
+ super.write(out);
out.writeUTF(mapredOutputFormat.getClass().getName());
jobConf.write(out);
}
@SuppressWarnings("unchecked")
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ super.read(in);
String hadoopOutputFormatName = in.readUTF();
if(jobConf == null) {
jobConf = new JobConf();
@@ -168,5 +180,11 @@ public abstract class HadoopOutputFormatBase<K, V, T> extends RichOutputFormat<T
throw new RuntimeException("Unable to instantiate the hadoop output format", e);
}
ReflectionUtils.setConf(mapredOutputFormat, jobConf);
+
+ jobConf.getCredentials().addAll(this.credentials);
+ Credentials currentUserCreds = getCredentialsFromUGI(UserGroupInformation.getCurrentUser());
+ if(currentUserCreds != null) {
+ jobConf.getCredentials().addAll(currentUserCreds);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormat.java
index 4d29d74..f0788c7 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormat.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormat.java
@@ -27,6 +27,12 @@ import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.hadoop.mapreduce.Job;
import org.apache.flink.api.java.typeutils.TypeExtractor;
+/**
+ * InputFormat implementation allowing to use Hadoop (mapreduce) InputFormats with Flink.
+ *
+ * @param <K> Key Type
+ * @param <V> Value Type
+ */
public class HadoopInputFormat<K, V> extends HadoopInputFormatBase<K, V, Tuple2<K, V>> implements ResultTypeQueryable<Tuple2<K,V>> {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java
index 3693176..09435e2 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopInputFormatBase.java
@@ -19,9 +19,9 @@
package org.apache.flink.api.java.hadoop.mapreduce;
import org.apache.flink.api.common.io.FileInputFormat.FileBaseStatistics;
-import org.apache.flink.api.common.io.RichInputFormat;
import org.apache.flink.api.common.io.LocatableInputSplitAssigner;
import org.apache.flink.api.common.io.statistics.BaseStatistics;
+import org.apache.flink.api.java.hadoop.common.HadoopInputFormatCommonBase;
import org.apache.flink.api.java.hadoop.mapreduce.utils.HadoopUtils;
import org.apache.flink.api.java.hadoop.mapreduce.wrapper.HadoopInputSplit;
import org.apache.flink.configuration.Configuration;
@@ -37,6 +37,8 @@ import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -48,12 +50,17 @@ import java.util.List;
import static com.google.common.base.Preconditions.checkNotNull;
-public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T, HadoopInputSplit> {
+/**
+ * Base class shared between the Java and Scala API of Flink
+ */
+public abstract class HadoopInputFormatBase<K, V, T> extends HadoopInputFormatCommonBase<T, HadoopInputSplit> {
private static final long serialVersionUID = 1L;
private static final Logger LOG = LoggerFactory.getLogger(HadoopInputFormatBase.class);
+ // NOTE: this class is using a custom serialization logic, without a defaultWriteObject() method.
+ // Hence, all fields here are "transient".
private org.apache.hadoop.mapreduce.InputFormat<K, V> mapreduceInputFormat;
protected Class<K> keyClass;
protected Class<V> valueClass;
@@ -64,11 +71,11 @@ public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T,
protected boolean hasNext;
public HadoopInputFormatBase(org.apache.hadoop.mapreduce.InputFormat<K, V> mapreduceInputFormat, Class<K> key, Class<V> value, Job job) {
- super();
+ super(checkNotNull(job, "Job can not be null").getCredentials());
this.mapreduceInputFormat = checkNotNull(mapreduceInputFormat);
this.keyClass = checkNotNull(key);
this.valueClass = checkNotNull(value);
- this.configuration = checkNotNull(job).getConfiguration();
+ this.configuration = job.getConfiguration();
HadoopUtils.mergeHadoopConf(configuration);
}
@@ -94,7 +101,7 @@ public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T,
return null;
}
- JobContext jobContext = null;
+ JobContext jobContext;
try {
jobContext = HadoopUtils.instantiateJobContext(configuration, null);
} catch (Exception e) {
@@ -104,23 +111,23 @@ public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T,
final FileBaseStatistics cachedFileStats = (cachedStats != null && cachedStats instanceof FileBaseStatistics) ?
(FileBaseStatistics) cachedStats : null;
- try {
- final org.apache.hadoop.fs.Path[] paths = FileInputFormat.getInputPaths(jobContext);
- return getFileStats(cachedFileStats, paths, new ArrayList<FileStatus>(1));
- } catch (IOException ioex) {
- if (LOG.isWarnEnabled()) {
- LOG.warn("Could not determine statistics due to an io error: "
- + ioex.getMessage());
- }
- } catch (Throwable t) {
- if (LOG.isErrorEnabled()) {
- LOG.error("Unexpected problem while getting the file statistics: "
- + t.getMessage(), t);
- }
- }
-
- // no statistics available
- return null;
+ try {
+ final org.apache.hadoop.fs.Path[] paths = FileInputFormat.getInputPaths(jobContext);
+ return getFileStats(cachedFileStats, paths, new ArrayList<FileStatus>(1));
+ } catch (IOException ioex) {
+ if (LOG.isWarnEnabled()) {
+ LOG.warn("Could not determine statistics due to an io error: "
+ + ioex.getMessage());
+ }
+ } catch (Throwable t) {
+ if (LOG.isErrorEnabled()) {
+ LOG.error("Unexpected problem while getting the file statistics: "
+ + t.getMessage(), t);
+ }
+ }
+
+ // no statistics available
+ return null;
}
@Override
@@ -128,13 +135,19 @@ public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T,
throws IOException {
configuration.setInt("mapreduce.input.fileinputformat.split.minsize", minNumSplits);
- JobContext jobContext = null;
+ JobContext jobContext;
try {
jobContext = HadoopUtils.instantiateJobContext(configuration, new JobID());
} catch (Exception e) {
throw new RuntimeException(e);
}
+ jobContext.getCredentials().addAll(this.credentials);
+ Credentials currentUserCreds = getCredentialsFromUGI(UserGroupInformation.getCurrentUser());
+ if(currentUserCreds != null) {
+ jobContext.getCredentials().addAll(currentUserCreds);
+ }
+
List<org.apache.hadoop.mapreduce.InputSplit> splits;
try {
splits = this.mapreduceInputFormat.getSplits(jobContext);
@@ -156,7 +169,7 @@ public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T,
@Override
public void open(HadoopInputSplit split) throws IOException {
- TaskAttemptContext context = null;
+ TaskAttemptContext context;
try {
context = HadoopUtils.instantiateTaskAttemptContext(configuration, new TaskAttemptID());
} catch(Exception e) {
@@ -255,6 +268,7 @@ public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T,
// --------------------------------------------------------------------------------------------
private void writeObject(ObjectOutputStream out) throws IOException {
+ super.write(out);
out.writeUTF(this.mapreduceInputFormat.getClass().getName());
out.writeUTF(this.keyClass.getName());
out.writeUTF(this.valueClass.getName());
@@ -263,6 +277,7 @@ public abstract class HadoopInputFormatBase<K, V, T> extends RichInputFormat<T,
@SuppressWarnings("unchecked")
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ super.read(in);
String hadoopInputFormatClassName = in.readUTF();
String keyClassName = in.readUTF();
String valueClassName = in.readUTF();
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormat.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormat.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormat.java
index 26a2c0d..36d8ad5 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormat.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormat.java
@@ -23,6 +23,12 @@ import java.io.IOException;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.hadoop.mapreduce.Job;
+/**
+ * OutputFormat implementation allowing to use Hadoop (mapreduce) OutputFormats with Flink.
+ *
+ * @param <K> Key Type
+ * @param <V> Value Type
+ */
public class HadoopOutputFormat<K, V> extends HadoopOutputFormatBase<K, V, Tuple2<K, V>> {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java
index 74ed6d0..72c105b 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/hadoop/mapreduce/HadoopOutputFormatBase.java
@@ -19,7 +19,7 @@
package org.apache.flink.api.java.hadoop.mapreduce;
import org.apache.flink.api.common.io.FinalizeOnMaster;
-import org.apache.flink.api.common.io.RichOutputFormat;
+import org.apache.flink.api.java.hadoop.common.HadoopOutputFormatCommonBase;
import org.apache.flink.api.java.hadoop.mapreduce.utils.HadoopUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.conf.Configurable;
@@ -33,13 +33,19 @@ import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import static org.apache.flink.api.java.hadoop.common.HadoopInputFormatCommonBase.getCredentialsFromUGI;
-public abstract class HadoopOutputFormatBase<K, V, T> extends RichOutputFormat<T> implements FinalizeOnMaster {
+/**
+ * Base class shared between the Java and Scala API of Flink
+ */
+public abstract class HadoopOutputFormatBase<K, V, T> extends HadoopOutputFormatCommonBase<T> implements FinalizeOnMaster {
private static final long serialVersionUID = 1L;
@@ -51,7 +57,7 @@ public abstract class HadoopOutputFormatBase<K, V, T> extends RichOutputFormat<T
private transient int taskNumber;
public HadoopOutputFormatBase(org.apache.hadoop.mapreduce.OutputFormat<K, V> mapreduceOutputFormat, Job job) {
- super();
+ super(job.getCredentials());
this.mapreduceOutputFormat = mapreduceOutputFormat;
this.configuration = job.getConfiguration();
HadoopUtils.mergeHadoopConf(configuration);
@@ -108,6 +114,12 @@ public abstract class HadoopOutputFormatBase<K, V, T> extends RichOutputFormat<T
throw new RuntimeException(e);
}
+ this.context.getCredentials().addAll(this.credentials);
+ Credentials currentUserCreds = getCredentialsFromUGI(UserGroupInformation.getCurrentUser());
+ if(currentUserCreds != null) {
+ this.context.getCredentials().addAll(currentUserCreds);
+ }
+
// compatible for hadoop 2.2.0, the temporary output directory is different from hadoop 1.2.1
if(outputCommitter instanceof FileOutputCommitter) {
this.configuration.set("mapreduce.task.output.dir", ((FileOutputCommitter)this.outputCommitter).getWorkPath().toString());
@@ -156,7 +168,6 @@ public abstract class HadoopOutputFormatBase<K, V, T> extends RichOutputFormat<T
JobContext jobContext;
TaskAttemptContext taskContext;
try {
-
TaskAttemptID taskAttemptID = TaskAttemptID.forName("attempt__0000_r_"
+ String.format("%" + (6 - Integer.toString(1).length()) + "s"," ").replace(" ", "0")
+ Integer.toString(1)
@@ -169,11 +180,16 @@ public abstract class HadoopOutputFormatBase<K, V, T> extends RichOutputFormat<T
throw new RuntimeException(e);
}
+ jobContext.getCredentials().addAll(this.credentials);
+ Credentials currentUserCreds = getCredentialsFromUGI(UserGroupInformation.getCurrentUser());
+ if(currentUserCreds != null) {
+ jobContext.getCredentials().addAll(currentUserCreds);
+ }
+
// finalize HDFS output format
if(this.outputCommitter != null) {
this.outputCommitter.commitJob(jobContext);
}
-
}
// --------------------------------------------------------------------------------------------
@@ -181,12 +197,14 @@ public abstract class HadoopOutputFormatBase<K, V, T> extends RichOutputFormat<T
// --------------------------------------------------------------------------------------------
private void writeObject(ObjectOutputStream out) throws IOException {
+ super.write(out);
out.writeUTF(this.mapreduceOutputFormat.getClass().getName());
this.configuration.write(out);
}
@SuppressWarnings("unchecked")
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ super.read(in);
String hadoopOutputFormatClassName = in.readUTF();
org.apache.hadoop.conf.Configuration configuration = new org.apache.hadoop.conf.Configuration();
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
index f3ca404..152b2e8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/fs/hdfs/HadoopFileSystem.java
@@ -441,7 +441,7 @@ public final class HadoopFileSystem extends FileSystem implements HadoopFileSyst
@Override
public Class<?> getHadoopWrapperClassNameForFileSystem(String scheme) {
Configuration hadoopConf = getHadoopConfiguration();
- Class<? extends org.apache.hadoop.fs.FileSystem> clazz = null;
+ Class<? extends org.apache.hadoop.fs.FileSystem> clazz;
// We can activate this block once we drop Hadoop1 support (only hd2 has the getFileSystemClass-method)
// try {
// clazz = org.apache.hadoop.fs.FileSystem.getFileSystemClass(scheme, hadoopConf);
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
index 17311e9..cdf7211 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
@@ -419,13 +419,13 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
* The given inputName is set on the given job.
*/
def readHadoopFile[K, V](
- mapredInputFormat: MapreduceFileInputFormat[K, V],
+ mapreduceInputFormat: MapreduceFileInputFormat[K, V],
key: Class[K],
value: Class[V],
inputPath: String,
job: Job)
(implicit tpe: TypeInformation[(K, V)]): DataSet[(K, V)] = {
- val result = createHadoopInput(mapredInputFormat, key, value, job)
+ val result = createHadoopInput(mapreduceInputFormat, key, value, job)
MapreduceFileInputFormat.addInputPath(job, new HadoopPath(inputPath))
result
}
@@ -436,25 +436,25 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
* [[org.apache.hadoop.mapreduce.Job]] with the given inputPath will be created.
*/
def readHadoopFile[K, V](
- mapredInputFormat: MapreduceFileInputFormat[K, V],
+ mapreduceInputFormat: MapreduceFileInputFormat[K, V],
key: Class[K],
value: Class[V],
inputPath: String)
(implicit tpe: TypeInformation[(K, V)]): DataSet[Tuple2[K, V]] = {
- readHadoopFile(mapredInputFormat, key, value, inputPath, Job.getInstance)
+ readHadoopFile(mapreduceInputFormat, key, value, inputPath, Job.getInstance)
}
/**
* Creates a [[DataSet]] from the given [[org.apache.hadoop.mapreduce.InputFormat]].
*/
def createHadoopInput[K, V](
- mapredInputFormat: MapreduceInputFormat[K, V],
+ mapreduceInputFormat: MapreduceInputFormat[K, V],
key: Class[K],
value: Class[V],
job: Job)
(implicit tpe: TypeInformation[(K, V)]): DataSet[Tuple2[K, V]] = {
val hadoopInputFormat =
- new mapreduce.HadoopInputFormat[K, V](mapredInputFormat, key, value, job)
+ new mapreduce.HadoopInputFormat[K, V](mapreduceInputFormat, key, value, job)
createInput(hadoopInputFormat)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-shaded-hadoop/pom.xml
----------------------------------------------------------------------
diff --git a/flink-shaded-hadoop/pom.xml b/flink-shaded-hadoop/pom.xml
index 8b4993e..d7e3158 100644
--- a/flink-shaded-hadoop/pom.xml
+++ b/flink-shaded-hadoop/pom.xml
@@ -61,7 +61,6 @@ under the License.
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
- <version>2.3</version>
<executions>
<execution>
<id>shade-hadoop</id>
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-flume/pom.xml
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-flume/pom.xml b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-flume/pom.xml
index 6bba57f..b4f5ad2 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-flume/pom.xml
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-flume/pom.xml
@@ -152,7 +152,6 @@ under the License.
<!-- Override artifactSet configuration to build fat-jar with all dependencies packed. -->
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
- <version>2.3</version>
<executions>
<execution>
<id>shade-flink</id>
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/pom.xml
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/pom.xml b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/pom.xml
index 9499ea5..f1ef0f0 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/pom.xml
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/pom.xml
@@ -74,7 +74,6 @@ under the License.
<!-- Override artifactSet configuration to build fat-jar with all dependencies packed. -->
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
- <version>2.3</version>
<executions>
<execution>
<id>shade-flink</id>
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/flink-yarn-tests/pom.xml
----------------------------------------------------------------------
diff --git a/flink-yarn-tests/pom.xml b/flink-yarn-tests/pom.xml
index b912fb0..2444ee4 100644
--- a/flink-yarn-tests/pom.xml
+++ b/flink-yarn-tests/pom.xml
@@ -130,7 +130,6 @@ under the License.
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
- <version>2.3</version>
<executions>
<execution>
<id>shade-hadoop</id>
http://git-wip-us.apache.org/repos/asf/flink/blob/b264b010/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 6af0355..b082162 100644
--- a/pom.xml
+++ b/pom.xml
@@ -906,7 +906,6 @@ under the License.
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
- <version>2.3</version>
<executions>
<execution>
<id>shade-flink</id>
@@ -962,6 +961,12 @@ under the License.
-->
<pluginManagement>
<plugins>
+ <!--set maven shade plugin version-->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <version>2.4.1</version>
+ </plugin>
<!--This plugin's configuration is used to store Eclipse m2e settings
only. It has no influence on the Maven build itself. -->
<plugin>