This is an automated email from the ASF dual-hosted git repository. jark 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 280df72 [FLINK-17968][hbase] Fix Hadoop Configuration is not properly serialized in HBaseRowInputFormat 280df72 is described below commit 280df7236a796d5b175f94952bea07b207ddf0d8 Author: zhangmang <zhangma...@163.com> AuthorDate: Tue Jun 2 11:26:32 2020 +0800 [FLINK-17968][hbase] Fix Hadoop Configuration is not properly serialized in HBaseRowInputFormat This closes #12146 --- .../flink/addons/hbase/TableInputFormat.java | 4 ++ .../hbase/source/AbstractTableInputFormat.java | 13 ++++ .../connector/hbase/source/HBaseInputFormat.java | 13 +++- .../hbase/source/HBaseRowDataInputFormat.java | 12 +--- .../hbase/source/HBaseRowInputFormat.java | 11 +-- .../connector/hbase/HBaseConnectorITCase.java | 12 ++-- .../connector/hbase/example/HBaseReadExample.java | 3 +- .../flink/connector/hbase/util/HBaseTestBase.java | 1 - .../hbase/util/HBaseTestingClusterAutoStarter.java | 83 +--------------------- 9 files changed, 40 insertions(+), 112 deletions(-) diff --git a/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/addons/hbase/TableInputFormat.java b/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/addons/hbase/TableInputFormat.java index 9be258e..a5e044e 100644 --- a/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/addons/hbase/TableInputFormat.java +++ b/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/addons/hbase/TableInputFormat.java @@ -31,4 +31,8 @@ import org.apache.flink.connector.hbase.source.HBaseInputFormat; public abstract class TableInputFormat<T extends Tuple> extends HBaseInputFormat<T> { private static final long serialVersionUID = 1L; + public TableInputFormat(org.apache.hadoop.conf.Configuration hConf) { + super(hConf); + } + } diff --git a/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/AbstractTableInputFormat.java b/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/AbstractTableInputFormat.java index 684afd2..94d36d3 100644 --- a/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/AbstractTableInputFormat.java +++ b/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/AbstractTableInputFormat.java @@ -24,8 +24,10 @@ import org.apache.flink.api.common.io.LocatableInputSplitAssigner; import org.apache.flink.api.common.io.RichInputFormat; import org.apache.flink.api.common.io.statistics.BaseStatistics; import org.apache.flink.configuration.Configuration; +import org.apache.flink.connector.hbase.util.HBaseConfigurationUtil; import org.apache.flink.core.io.InputSplitAssigner; +import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.ResultScanner; @@ -60,6 +62,13 @@ abstract class AbstractTableInputFormat<T> extends RichInputFormat<T, TableInput protected byte[] currentRow; protected long scannedRows; + // Configuration is not serializable + protected byte[] serializedConfig; + + public AbstractTableInputFormat(org.apache.hadoop.conf.Configuration hConf) { + serializedConfig = HBaseConfigurationUtil.serializeConfiguration(hConf); + } + /** * Returns an instance of Scan that retrieves the required subset of records from the HBase table. * @@ -99,6 +108,10 @@ abstract class AbstractTableInputFormat<T> extends RichInputFormat<T, TableInput */ public abstract void configure(Configuration parameters); + protected org.apache.hadoop.conf.Configuration getHadoopConfiguration() { + return HBaseConfigurationUtil.deserializeConfiguration(serializedConfig, HBaseConfiguration.create()); + } + @Override public void open(TableInputSplit split) throws IOException { if (table == null) { diff --git a/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseInputFormat.java b/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseInputFormat.java index 972cf0e..45d5b3f 100644 --- a/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseInputFormat.java +++ b/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseInputFormat.java @@ -23,7 +23,6 @@ import org.apache.flink.api.common.io.InputFormat; import org.apache.flink.api.java.tuple.Tuple; import org.apache.flink.configuration.Configuration; -import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.Scan; @@ -37,6 +36,15 @@ public abstract class HBaseInputFormat<T extends Tuple> extends AbstractTableInp private static final long serialVersionUID = 1L; /** + * Constructs a {@link InputFormat} with hbase configuration to read data from hbase. + * @param hConf The configuration that connect to hbase. + * At least hbase.zookeeper.quorum and zookeeper.znode.parent need to be set. + */ + public HBaseInputFormat(org.apache.hadoop.conf.Configuration hConf) { + super(hConf); + } + + /** * Returns an instance of Scan that retrieves the required subset of records from the HBase table. * @return The appropriate instance of Scan for this usecase. */ @@ -79,8 +87,7 @@ public abstract class HBaseInputFormat<T extends Tuple> extends AbstractTableInp */ private HTable createTable() { LOG.info("Initializing HBaseConfiguration"); - //use files found in the classpath - org.apache.hadoop.conf.Configuration hConf = HBaseConfiguration.create(); + org.apache.hadoop.conf.Configuration hConf = getHadoopConfiguration(); try { return new HTable(hConf, getTableName()); diff --git a/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseRowDataInputFormat.java b/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseRowDataInputFormat.java index 5a628d9..30be6d8 100644 --- a/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseRowDataInputFormat.java +++ b/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseRowDataInputFormat.java @@ -24,7 +24,6 @@ import org.apache.flink.connector.hbase.util.HBaseSerde; import org.apache.flink.connector.hbase.util.HBaseTableSchema; import org.apache.flink.table.data.RowData; -import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.TableNotFoundException; import org.apache.hadoop.hbase.client.Connection; @@ -50,15 +49,13 @@ public class HBaseRowDataInputFormat extends AbstractTableInputFormat<RowData> { private transient HBaseSerde serde; - private transient org.apache.hadoop.conf.Configuration conf; - public HBaseRowDataInputFormat( org.apache.hadoop.conf.Configuration conf, String tableName, HBaseTableSchema schema, String nullStringLiteral) { + super(conf); this.tableName = tableName; - this.conf = conf; this.schema = schema; this.nullStringLiteral = nullStringLiteral; } @@ -89,13 +86,8 @@ public class HBaseRowDataInputFormat extends AbstractTableInputFormat<RowData> { } private void connectToTable() { - - if (this.conf == null) { - this.conf = HBaseConfiguration.create(); - } - try { - Connection conn = ConnectionFactory.createConnection(conf); + Connection conn = ConnectionFactory.createConnection(getHadoopConfiguration()); super.table = (HTable) conn.getTable(TableName.valueOf(tableName)); } catch (TableNotFoundException tnfe) { LOG.error("The table " + tableName + " not found ", tnfe); diff --git a/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseRowInputFormat.java b/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseRowInputFormat.java index f455298..f7100ed 100644 --- a/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseRowInputFormat.java +++ b/flink-connectors/flink-connector-hbase/src/main/java/org/apache/flink/connector/hbase/source/HBaseRowInputFormat.java @@ -28,7 +28,6 @@ import org.apache.flink.connector.hbase.util.HBaseReadWriteHelper; import org.apache.flink.connector.hbase.util.HBaseTableSchema; import org.apache.flink.types.Row; -import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.TableNotFoundException; import org.apache.hadoop.hbase.client.Connection; @@ -54,12 +53,11 @@ public class HBaseRowInputFormat extends AbstractTableInputFormat<Row> implement private final String tableName; private final HBaseTableSchema schema; - private transient org.apache.hadoop.conf.Configuration conf; private transient HBaseReadWriteHelper readHelper; public HBaseRowInputFormat(org.apache.hadoop.conf.Configuration conf, String tableName, HBaseTableSchema schema) { + super(conf); this.tableName = tableName; - this.conf = conf; this.schema = schema; } @@ -90,13 +88,8 @@ public class HBaseRowInputFormat extends AbstractTableInputFormat<Row> implement } private void connectToTable() { - - if (this.conf == null) { - this.conf = HBaseConfiguration.create(); - } - try { - Connection conn = ConnectionFactory.createConnection(conf); + Connection conn = ConnectionFactory.createConnection(getHadoopConfiguration()); super.table = (HTable) conn.getTable(TableName.valueOf(tableName)); } catch (TableNotFoundException tnfe) { LOG.error("The table " + tableName + " not found ", tnfe); diff --git a/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/HBaseConnectorITCase.java b/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/HBaseConnectorITCase.java index 2a1ffa2..7724873 100644 --- a/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/HBaseConnectorITCase.java +++ b/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/HBaseConnectorITCase.java @@ -54,8 +54,6 @@ import org.apache.flink.table.sources.TableSource; import org.apache.flink.test.util.TestBaseUtils; import org.apache.flink.types.Row; -import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.util.Bytes; @@ -312,7 +310,7 @@ public class HBaseConnectorITCase extends HBaseTestBase { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); DataSet<Tuple1<Integer>> result = env - .createInput(new InputFormatForTestTable()) + .createInput(new InputFormatForTestTable(getConf())) .reduce((ReduceFunction<Tuple1<Integer>>) (v1, v2) -> Tuple1.of(v1.f0 + v2.f0)); List<Tuple1<Integer>> resultSet = result.collect(); @@ -612,9 +610,7 @@ public class HBaseConnectorITCase extends HBaseTestBase { properties.put(CONNECTOR_VERSION, CONNECTOR_VERSION_VALUE_143); properties.put(CONNECTOR_PROPERTY_VERSION, "1"); properties.put(CONNECTOR_TABLE_NAME, TEST_TABLE_1); - // get zk quorum from "hbase-site.xml" in classpath - String hbaseZk = HBaseConfiguration.create().get(HConstants.ZOOKEEPER_QUORUM); - properties.put(CONNECTOR_ZK_QUORUM, hbaseZk); + properties.put(CONNECTOR_ZK_QUORUM, getZookeeperQuorum()); // schema String[] columnNames = {FAMILY1, ROWKEY, FAMILY2, FAMILY3}; TypeInformation<Row> f1 = Types.ROW_NAMED(new String[]{F1COL1}, Types.INT); @@ -705,6 +701,10 @@ public class HBaseConnectorITCase extends HBaseTestBase { public static class InputFormatForTestTable extends HBaseInputFormat<Tuple1<Integer>> { private static final long serialVersionUID = 1L; + public InputFormatForTestTable(org.apache.hadoop.conf.Configuration hConf) { + super(hConf); + } + @Override protected Scan getScanner() { return new Scan(); diff --git a/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/example/HBaseReadExample.java b/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/example/HBaseReadExample.java index 13914e5..026c6d0 100644 --- a/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/example/HBaseReadExample.java +++ b/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/example/HBaseReadExample.java @@ -24,6 +24,7 @@ import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.connector.hbase.source.HBaseInputFormat; +import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.util.Bytes; @@ -47,7 +48,7 @@ public class HBaseReadExample { public static void main(String[] args) throws Exception { ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); @SuppressWarnings("serial") - DataSet<Tuple2<String, String>> hbaseDs = env.createInput(new HBaseInputFormat<Tuple2<String, String>>() { + DataSet<Tuple2<String, String>> hbaseDs = env.createInput(new HBaseInputFormat<Tuple2<String, String>>(HBaseConfiguration.create()) { @Override public String getTableName() { diff --git a/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/util/HBaseTestBase.java b/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/util/HBaseTestBase.java index 04555cc..e00b986 100644 --- a/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/util/HBaseTestBase.java +++ b/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/util/HBaseTestBase.java @@ -74,7 +74,6 @@ public abstract class HBaseTestBase extends HBaseTestingClusterAutoStarter { @BeforeClass public static void activateHBaseCluster() throws IOException { - registerHBaseMiniClusterInClasspath(); prepareTables(); } diff --git a/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/util/HBaseTestingClusterAutoStarter.java b/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/util/HBaseTestingClusterAutoStarter.java index 9e15a2f..4ba1c17 100644 --- a/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/util/HBaseTestingClusterAutoStarter.java +++ b/flink-connectors/flink-connector-hbase/src/test/java/org/apache/flink/connector/hbase/util/HBaseTestingClusterAutoStarter.java @@ -40,24 +40,13 @@ import org.junit.AfterClass; import org.junit.Assume; import org.junit.BeforeClass; -import java.io.File; -import java.io.FileOutputStream; import java.io.IOException; -import java.io.OutputStream; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; -import java.net.MalformedURLException; -import java.net.URL; -import java.net.URLClassLoader; -import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; -import java.util.Random; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; /** * By using this class as the super class of a set of tests you will have a HBase testing @@ -83,8 +72,6 @@ public abstract class HBaseTestingClusterAutoStarter extends AbstractTestBase { private static HBaseAdmin admin = null; private static List<TableName> createdTables = new ArrayList<>(); - private static boolean alreadyRegisteredTestCluster = false; - private static Configuration conf; protected static void createTable(TableName tableName, byte[][] columnFamilyName, byte[][] splitKeys) { @@ -157,6 +144,7 @@ public abstract class HBaseTestingClusterAutoStarter extends AbstractTestBase { TEST_UTIL.getConfiguration().setInt("hbase.master.info.port", -1); // Make sure the zookeeper quorum value contains the right port number (varies per run). + LOG.info("Hbase minicluster client port: " + TEST_UTIL.getZkCluster().getClientPort()); TEST_UTIL.getConfiguration().set("hbase.zookeeper.quorum", "localhost:" + TEST_UTIL.getZkCluster().getClientPort()); conf = initialize(TEST_UTIL.getConfiguration()); @@ -170,77 +158,10 @@ public abstract class HBaseTestingClusterAutoStarter extends AbstractTestBase { return "localhost:" + TEST_UTIL.getZkCluster().getClientPort(); } - private static File hbaseSiteXmlDirectory; - private static File hbaseSiteXmlFile; - - /** - * This dynamically generates a hbase-site.xml file that is added to the classpath. - * This way this HBaseMinicluster can be used by an unmodified application. - * The downside is that this cannot be 'unloaded' so you can have only one per JVM. - */ - public static void registerHBaseMiniClusterInClasspath() { - if (alreadyRegisteredTestCluster) { - fail("You CANNOT register a second HBase Testing cluster in the classpath of the SAME JVM"); - } - File baseDir = new File(System.getProperty("java.io.tmpdir", "/tmp/")); - hbaseSiteXmlDirectory = new File(baseDir, "unittest-hbase-minicluster-" + Math.abs(new Random().nextLong()) + "/"); - - if (!hbaseSiteXmlDirectory.mkdirs()) { - fail("Unable to create output directory " + hbaseSiteXmlDirectory + " for the HBase minicluster"); - } - - assertNotNull("The ZooKeeper for the HBase minicluster is missing", TEST_UTIL.getZkCluster()); - - createHBaseSiteXml(hbaseSiteXmlDirectory, TEST_UTIL.getConfiguration().get("hbase.zookeeper.quorum")); - addDirectoryToClassPath(hbaseSiteXmlDirectory); - - // Avoid starting it again. - alreadyRegisteredTestCluster = true; - } - public static Configuration getConf() { return conf; } - private static void createHBaseSiteXml(File hbaseSiteXmlDirectory, String zookeeperQuorum) { - hbaseSiteXmlFile = new File(hbaseSiteXmlDirectory, "hbase-site.xml"); - // Create the hbase-site.xml file for this run. - try { - String hbaseSiteXml = "<?xml version=\"1.0\"?>\n" + - "<?xml-stylesheet type=\"text/xsl\" href=\"configuration.xsl\"?>\n" + - "<configuration>\n" + - " <property>\n" + - " <name>hbase.zookeeper.quorum</name>\n" + - " <value>" + zookeeperQuorum + "</value>\n" + - " </property>\n" + - "</configuration>"; - OutputStream fos = new FileOutputStream(hbaseSiteXmlFile); - fos.write(hbaseSiteXml.getBytes(StandardCharsets.UTF_8)); - fos.close(); - } catch (IOException e) { - fail("Unable to create " + hbaseSiteXmlFile); - } - } - - private static void addDirectoryToClassPath(File directory) { - try { - // Get the classloader actually used by HBaseConfiguration - ClassLoader classLoader = HBaseConfiguration.create().getClassLoader(); - if (!(classLoader instanceof URLClassLoader)) { - fail("We should get a URLClassLoader"); - } - - // Make the addURL method accessible - Method method = URLClassLoader.class.getDeclaredMethod("addURL", URL.class); - method.setAccessible(true); - - // Add the directory where we put the hbase-site.xml to the classpath - method.invoke(classLoader, directory.toURI().toURL()); - } catch (MalformedURLException | NoSuchMethodException | IllegalAccessException | InvocationTargetException e) { - fail("Unable to add " + directory + " to classpath because of this exception: " + e.getMessage()); - } - } - @AfterClass public static void tearDown() throws Exception { if (conf == null) { @@ -249,8 +170,6 @@ public abstract class HBaseTestingClusterAutoStarter extends AbstractTestBase { } LOG.info("HBase minicluster: Shutting down"); deleteTables(); - hbaseSiteXmlFile.delete(); - hbaseSiteXmlDirectory.delete(); TEST_UTIL.shutdownMiniCluster(); LOG.info("HBase minicluster: Down"); }