[flink] branch release-1.11 updated: [FLINK-19398][connectors/hive] Fix the failure when creating hive connector from userclassloader

[lz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.11 by this push:
     new 4616f7d  [FLINK-19398][connectors/hive] Fix the failure when creating hive connector from userclassloader
4616f7d is described below

commit 4616f7d954275081a9a97c5745d5f55072241d17
Author: Yun Gao <ga...@gmail.com>
AuthorDate: Thu Dec 3 16:11:35 2020 +0800

    [FLINK-19398][connectors/hive] Fix the failure when creating hive connector from userclassloader
    
    This closes #14286
---
 .../bulk/HadoopPathBasedPartFileWriterTest.java    | 53 ------------
 .../bulk/TestHadoopPathBasedBulkWriterFactory.java | 82 +++++++++++++++++++
 .../HadoopPathBasedBulkFormatBuilderTest.java      | 95 ++++++++++++++++++++++
 .../functions/sink/filesystem/BucketFactory.java   |  2 +-
 .../api/functions/sink/filesystem/Buckets.java     |  2 +-
 .../sink/filesystem/DefaultBucketFactoryImpl.java  |  2 +-
 6 files changed, 180 insertions(+), 56 deletions(-)

diff --git a/flink-formats/flink-hadoop-bulk/src/test/java/org/apache/flink/formats/hadoop/bulk/HadoopPathBasedPartFileWriterTest.java b/flink-formats/flink-hadoop-bulk/src/test/java/org/apache/flink/formats/hadoop/bulk/HadoopPathBasedPartFileWriterTest.java
index 3ee14ee..c0d337a 100644
--- a/flink-formats/flink-hadoop-bulk/src/test/java/org/apache/flink/formats/hadoop/bulk/HadoopPathBasedPartFileWriterTest.java
+++ b/flink-formats/flink-hadoop-bulk/src/test/java/org/apache/flink/formats/hadoop/bulk/HadoopPathBasedPartFileWriterTest.java
@@ -26,12 +26,9 @@ import org.apache.flink.streaming.api.functions.sink.filesystem.TestStreamingFil
 import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
 import org.apache.flink.streaming.util.FiniteTestSource;
 import org.apache.flink.test.util.AbstractTestBase;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.IOUtils;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -140,54 +137,4 @@ public class HadoopPathBasedPartFileWriterTest extends AbstractTestBase {
 			return lines;
 		}
 	}
-
-	private static class TestHadoopPathBasedBulkWriterFactory implements HadoopPathBasedBulkWriter.Factory<String> {
-
-		@Override
-		public HadoopPathBasedBulkWriter<String> create(Path targetFilePath, Path inProgressFilePath) {
-			try {
-				FileSystem fileSystem = FileSystem.get(inProgressFilePath.toUri(), new Configuration());
-				FSDataOutputStream output = fileSystem.create(inProgressFilePath);
-				return new FSDataOutputStreamBulkWriterHadoop(output);
-			} catch (IOException e) {
-				ExceptionUtils.rethrow(e);
-			}
-
-			return null;
-		}
-	}
-
-	private static class FSDataOutputStreamBulkWriterHadoop implements HadoopPathBasedBulkWriter<String> {
-		private final FSDataOutputStream outputStream;
-
-		public FSDataOutputStreamBulkWriterHadoop(FSDataOutputStream outputStream) {
-			this.outputStream = outputStream;
-		}
-
-		@Override
-		public long getSize() throws IOException {
-			return outputStream.getPos();
-		}
-
-		@Override
-		public void dispose() {
-			IOUtils.closeQuietly(outputStream);
-		}
-
-		@Override
-		public void addElement(String element) throws IOException {
-			outputStream.writeBytes(element + "\n");
-		}
-
-		@Override
-		public void flush() throws IOException {
-			outputStream.flush();
-		}
-
-		@Override
-		public void finish() throws IOException {
-			outputStream.flush();
-			outputStream.close();
-		}
-	}
 }
diff --git a/flink-formats/flink-hadoop-bulk/src/test/java/org/apache/flink/formats/hadoop/bulk/TestHadoopPathBasedBulkWriterFactory.java b/flink-formats/flink-hadoop-bulk/src/test/java/org/apache/flink/formats/hadoop/bulk/TestHadoopPathBasedBulkWriterFactory.java
new file mode 100644
index 0000000..17b2aa9
--- /dev/null
+++ b/flink-formats/flink-hadoop-bulk/src/test/java/org/apache/flink/formats/hadoop/bulk/TestHadoopPathBasedBulkWriterFactory.java
@@ -0,0 +1,82 @@
+/*
+ * 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.formats.hadoop.bulk;
+
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.IOUtils;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+
+/**
+ * A {@link HadoopPathBasedBulkWriter.Factory} implementation used in tests.
+ */
+public class TestHadoopPathBasedBulkWriterFactory implements HadoopPathBasedBulkWriter.Factory<String> {
+
+	@Override
+	public HadoopPathBasedBulkWriter<String> create(Path targetFilePath, Path inProgressFilePath) {
+		try {
+			FileSystem fileSystem = FileSystem.get(inProgressFilePath.toUri(), new Configuration());
+			FSDataOutputStream output = fileSystem.create(inProgressFilePath);
+			return new FSDataOutputStreamBulkWriterHadoop(output);
+		} catch (IOException e) {
+			ExceptionUtils.rethrow(e);
+		}
+
+		return null;
+	}
+
+	private static class FSDataOutputStreamBulkWriterHadoop implements HadoopPathBasedBulkWriter<String> {
+		private final FSDataOutputStream outputStream;
+
+		public FSDataOutputStreamBulkWriterHadoop(FSDataOutputStream outputStream) {
+			this.outputStream = outputStream;
+		}
+
+		@Override
+		public long getSize() throws IOException {
+			return outputStream.getPos();
+		}
+
+		@Override
+		public void dispose() {
+			IOUtils.closeQuietly(outputStream);
+		}
+
+		@Override
+		public void addElement(String element) throws IOException {
+			outputStream.writeBytes(element + "\n");
+		}
+
+		@Override
+		public void flush() throws IOException {
+			outputStream.flush();
+		}
+
+		@Override
+		public void finish() throws IOException {
+			outputStream.flush();
+			outputStream.close();
+		}
+	}
+}
diff --git a/flink-formats/flink-hadoop-bulk/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/HadoopPathBasedBulkFormatBuilderTest.java b/flink-formats/flink-hadoop-bulk/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/HadoopPathBasedBulkFormatBuilderTest.java
new file mode 100644
index 0000000..1ff8ce2
--- /dev/null
+++ b/flink-formats/flink-hadoop-bulk/src/test/java/org/apache/flink/streaming/api/functions/sink/filesystem/HadoopPathBasedBulkFormatBuilderTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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.streaming.api.functions.sink.filesystem;
+
+import org.apache.flink.formats.hadoop.bulk.HadoopPathBasedBulkWriter;
+import org.apache.flink.formats.hadoop.bulk.TestHadoopPathBasedBulkWriterFactory;
+import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner;
+import org.apache.flink.util.FlinkUserCodeClassLoader;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.lang.reflect.Constructor;
+import java.net.URL;
+import java.net.URLClassLoader;
+
+import static org.junit.Assert.assertNotNull;
+
+/**
+ * Tests the behaviors of {@link HadoopPathBasedBulkFormatBuilder}.
+ */
+public class HadoopPathBasedBulkFormatBuilderTest {
+
+	/**
+	 * Tests if we could create {@link HadoopPathBasedBulkFormatBuilder} within user classloader.
+	 * It is mainly verify we have fixed the issue raised in https://issues.apache.org/jira/browse/FLINK-19398.
+	 */
+	@Test
+	@SuppressWarnings({"unchecked", "rawtypes"})
+	public void testCreatingBuildWithinUserClassLoader() throws Exception {
+		ClassLoader appClassLoader = getClass().getClassLoader();
+		Assume.assumeTrue(appClassLoader instanceof URLClassLoader);
+
+		ClassLoader userClassLoader = new SpecifiedChildFirstUserClassLoader(
+				HadoopPathBasedBulkFormatBuilder.class.getName(),
+				appClassLoader,
+				((URLClassLoader) appClassLoader).getURLs());
+
+		Class<HadoopPathBasedBulkFormatBuilder> userHadoopFormatBuildClass =
+				(Class<HadoopPathBasedBulkFormatBuilder>) userClassLoader.loadClass(
+						HadoopPathBasedBulkFormatBuilder.class.getName());
+		Constructor<?> constructor = userHadoopFormatBuildClass.getConstructor(
+				Path.class,
+				HadoopPathBasedBulkWriter.Factory.class,
+				Configuration.class,
+				BucketAssigner.class);
+		Object hadoopFormatBuilder = constructor.newInstance(
+				new Path("/tmp"),
+				new TestHadoopPathBasedBulkWriterFactory(),
+				new Configuration(),
+				new DateTimeBucketAssigner<>());
+
+		Buckets<String, String> buckets = (Buckets<String, String>) userHadoopFormatBuildClass
+				.getMethod("createBuckets", int.class)
+				.invoke(hadoopFormatBuilder, 0);
+		assertNotNull(buckets);
+	}
+
+	private static class SpecifiedChildFirstUserClassLoader extends FlinkUserCodeClassLoader {
+
+		private final String specifiedClassName;
+
+		protected SpecifiedChildFirstUserClassLoader(String specifiedClassName, ClassLoader parent, URL[] urls) {
+			super(urls, parent);
+			this.specifiedClassName = specifiedClassName;
+		}
+
+		@Override
+		protected Class<?> loadClassWithoutExceptionHandling(String name, boolean resolve) throws ClassNotFoundException {
+			if (name.equals(specifiedClassName)) {
+				return findClass(name);
+			} else {
+				return super.loadClassWithoutExceptionHandling(name, resolve);
+			}
+		}
+	}
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketFactory.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketFactory.java
index 6423627..bc51f21 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketFactory.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/BucketFactory.java
@@ -28,7 +28,7 @@ import java.io.Serializable;
  * A factory able to create {@link Bucket buckets} for the {@link StreamingFileSink}.
  */
 @Internal
-interface BucketFactory<IN, BucketID> extends Serializable {
+public interface BucketFactory<IN, BucketID> extends Serializable {
 
 	Bucket<IN, BucketID> getNewBucket(
 			final int subtaskIndex,
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Buckets.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Buckets.java
index f936348..eeed2c6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Buckets.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/Buckets.java
@@ -91,7 +91,7 @@ public class Buckets<IN, BucketID> {
 	 * @param bucketWriter The {@link BucketWriter} to be used when writing data.
 	 * @param rollingPolicy The {@link RollingPolicy} as specified by the user.
 	 */
-	Buckets(
+	public Buckets(
 			final Path basePath,
 			final BucketAssigner<IN, BucketID> bucketAssigner,
 			final BucketFactory<IN, BucketID> bucketFactory,
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/DefaultBucketFactoryImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/DefaultBucketFactoryImpl.java
index bb20b97..be992ed 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/DefaultBucketFactoryImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/filesystem/DefaultBucketFactoryImpl.java
@@ -27,7 +27,7 @@ import java.io.IOException;
  * A factory returning {@link Bucket buckets}.
  */
 @Internal
-class DefaultBucketFactoryImpl<IN, BucketID> implements BucketFactory<IN, BucketID> {
+public class DefaultBucketFactoryImpl<IN, BucketID> implements BucketFactory<IN, BucketID> {
 
 	private static final long serialVersionUID = 1L;