sankarh commented on a change in pull request #1146: URL: https://github.com/apache/hive/pull/1146#discussion_r444398139
########## File path: ql/src/java/org/apache/hadoop/hive/ql/reexec/ReExecuteLostAMQueryPlugin.java ########## @@ -0,0 +1,75 @@ +/* + * 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.hadoop.hive.ql.reexec; + +import org.apache.hadoop.hive.ql.Driver; +import org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext; +import org.apache.hadoop.hive.ql.hooks.HookContext; +import org.apache.hadoop.hive.ql.plan.mapper.PlanMapper; + +import java.util.regex.Pattern; + +public class ReExecuteLostAMQueryPlugin implements IReExecutionPlugin { + private boolean retryPossible; Review comment: In Hive code, we follow 2 spaced tabs. Pls update it in newly created files in this patch. ########## File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/reexec/TestReExecuteKilledTezAMQueryPlugin.java ########## @@ -0,0 +1,207 @@ +/* + * 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.hadoop.hive.ql.reexec; + +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.LlapBaseInputFormat; +import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.yarn.api.records.ApplicationReport; +import org.apache.hadoop.yarn.api.records.YarnApplicationState; +import org.apache.hadoop.yarn.client.api.YarnClient; +import org.apache.hive.jdbc.BaseJdbcWithMiniLlap; +import org.apache.hive.jdbc.HiveStatement; +import org.apache.hive.jdbc.TestJdbcWithMiniLlapArrow; +import org.apache.hive.jdbc.miniHS2.MiniHS2; +import org.junit.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.net.URL; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + + +public class TestReExecuteKilledTezAMQueryPlugin { + protected static final Logger LOG = LoggerFactory.getLogger(TestJdbcWithMiniLlapArrow.class); + + private static MiniHS2 miniHS2 = null; + private static final String tableName = "testKillTezAmTbl"; + private static String dataFileDir; + private static final String testDbName = "testKillTezAmDb"; + protected static Connection hs2Conn = null; + private static HiveConf conf; + + private static class ExceptionHolder { + Throwable throwable; + } + + static HiveConf defaultConf() throws Exception { + String confDir = "../../data/conf/llap/"; + if (confDir != null && !confDir.isEmpty()) { + HiveConf.setHiveSiteLocation(new URL("file://"+ new File(confDir).toURI().getPath() + "/hive-site.xml")); + System.out.println("Setting hive-site: " + HiveConf.getHiveSiteLocation()); + } + HiveConf defaultConf = new HiveConf(); + defaultConf.setBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY, false); + defaultConf.setBoolVar(HiveConf.ConfVars.HIVE_SERVER2_ENABLE_DOAS, false); + defaultConf.addResource(new URL("file://" + new File(confDir).toURI().getPath() + "/tez-site.xml")); + return defaultConf; + } + + @BeforeClass + public static void beforeTest() throws Exception { + conf = defaultConf(); + conf.setVar(HiveConf.ConfVars.USERS_IN_ADMIN_ROLE, System.getProperty("user.name")); + conf.set(HiveConf.ConfVars.HIVE_QUERY_REEXECUTION_STRATEGIES.varname, "reexecute_lost_am"); + MiniHS2.cleanupLocalDir(); + Class.forName(MiniHS2.getJdbcDriverName()); + miniHS2 = new MiniHS2(conf, MiniHS2.MiniClusterType.LLAP); + dataFileDir = conf.get("test.data.files").replace('\\', '/').replace("c:", ""); + Map<String, String> confOverlay = new HashMap<String, String>(); + miniHS2.start(confOverlay); + miniHS2.getDFS().getFileSystem().mkdirs(new Path("/apps_staging_dir/anonymous")); + + Connection conDefault = getConnection(miniHS2.getJdbcURL(), + System.getProperty("user.name"), "bar"); + Statement stmt = conDefault.createStatement(); + String tblName = testDbName + "." + tableName; + Path dataFilePath = new Path(dataFileDir, "kv1.txt"); + String udfName = TestJdbcWithMiniLlapArrow.SleepMsUDF.class.getName(); + stmt.execute("drop database if exists " + testDbName + " cascade"); + stmt.execute("create database " + testDbName); + stmt.execute("set role admin"); + stmt.execute("dfs -put " + dataFilePath.toString() + " " + "kv1.txt"); + stmt.execute("use " + testDbName); + stmt.execute("create table " + tblName + " (int_col int, value string) "); + stmt.execute("load data inpath 'kv1.txt' into table " + tblName); + stmt.execute("create function sleepMsUDF as '" + udfName + "'"); + stmt.execute("grant select on table " + tblName + " to role public"); + + stmt.close(); + conDefault.close(); + } + + @AfterClass + public static void afterTest() { + if (miniHS2 != null && miniHS2.isStarted()) { + miniHS2.stop(); + } + } + + @Before + public void setUp() throws Exception { + hs2Conn = getConnection(miniHS2.getJdbcURL(), System.getProperty("user.name"), "bar"); + } + + public static Connection getConnection(String jdbcURL, String user, String pwd) throws SQLException { + Connection conn = DriverManager.getConnection(jdbcURL, user, pwd); + conn.createStatement().execute("set hive.support.concurrency = false"); + return conn; + } + + @After + public void tearDown() throws Exception { + LlapBaseInputFormat.closeAll(); + hs2Conn.close(); + } + Review comment: Too many blank lines. ########## File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/reexec/TestReExecuteKilledTezAMQueryPlugin.java ########## @@ -0,0 +1,207 @@ +/* + * 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.hadoop.hive.ql.reexec; + +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.LlapBaseInputFormat; +import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.yarn.api.records.ApplicationReport; +import org.apache.hadoop.yarn.api.records.YarnApplicationState; +import org.apache.hadoop.yarn.client.api.YarnClient; +import org.apache.hive.jdbc.BaseJdbcWithMiniLlap; +import org.apache.hive.jdbc.HiveStatement; +import org.apache.hive.jdbc.TestJdbcWithMiniLlapArrow; +import org.apache.hive.jdbc.miniHS2.MiniHS2; +import org.junit.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.net.URL; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + + +public class TestReExecuteKilledTezAMQueryPlugin { + protected static final Logger LOG = LoggerFactory.getLogger(TestJdbcWithMiniLlapArrow.class); + + private static MiniHS2 miniHS2 = null; + private static final String tableName = "testKillTezAmTbl"; + private static String dataFileDir; + private static final String testDbName = "testKillTezAmDb"; + protected static Connection hs2Conn = null; + private static HiveConf conf; + + private static class ExceptionHolder { + Throwable throwable; + } + + static HiveConf defaultConf() throws Exception { + String confDir = "../../data/conf/llap/"; + if (confDir != null && !confDir.isEmpty()) { + HiveConf.setHiveSiteLocation(new URL("file://"+ new File(confDir).toURI().getPath() + "/hive-site.xml")); + System.out.println("Setting hive-site: " + HiveConf.getHiveSiteLocation()); + } + HiveConf defaultConf = new HiveConf(); + defaultConf.setBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY, false); + defaultConf.setBoolVar(HiveConf.ConfVars.HIVE_SERVER2_ENABLE_DOAS, false); + defaultConf.addResource(new URL("file://" + new File(confDir).toURI().getPath() + "/tez-site.xml")); + return defaultConf; + } + + @BeforeClass + public static void beforeTest() throws Exception { + conf = defaultConf(); + conf.setVar(HiveConf.ConfVars.USERS_IN_ADMIN_ROLE, System.getProperty("user.name")); + conf.set(HiveConf.ConfVars.HIVE_QUERY_REEXECUTION_STRATEGIES.varname, "reexecute_lost_am"); + MiniHS2.cleanupLocalDir(); + Class.forName(MiniHS2.getJdbcDriverName()); + miniHS2 = new MiniHS2(conf, MiniHS2.MiniClusterType.LLAP); + dataFileDir = conf.get("test.data.files").replace('\\', '/').replace("c:", ""); + Map<String, String> confOverlay = new HashMap<String, String>(); + miniHS2.start(confOverlay); + miniHS2.getDFS().getFileSystem().mkdirs(new Path("/apps_staging_dir/anonymous")); + + Connection conDefault = getConnection(miniHS2.getJdbcURL(), + System.getProperty("user.name"), "bar"); + Statement stmt = conDefault.createStatement(); + String tblName = testDbName + "." + tableName; + Path dataFilePath = new Path(dataFileDir, "kv1.txt"); + String udfName = TestJdbcWithMiniLlapArrow.SleepMsUDF.class.getName(); + stmt.execute("drop database if exists " + testDbName + " cascade"); + stmt.execute("create database " + testDbName); + stmt.execute("set role admin"); + stmt.execute("dfs -put " + dataFilePath.toString() + " " + "kv1.txt"); + stmt.execute("use " + testDbName); + stmt.execute("create table " + tblName + " (int_col int, value string) "); + stmt.execute("load data inpath 'kv1.txt' into table " + tblName); + stmt.execute("create function sleepMsUDF as '" + udfName + "'"); + stmt.execute("grant select on table " + tblName + " to role public"); + + stmt.close(); + conDefault.close(); + } + + @AfterClass + public static void afterTest() { + if (miniHS2 != null && miniHS2.isStarted()) { + miniHS2.stop(); + } + } + + @Before + public void setUp() throws Exception { + hs2Conn = getConnection(miniHS2.getJdbcURL(), System.getProperty("user.name"), "bar"); + } + + public static Connection getConnection(String jdbcURL, String user, String pwd) throws SQLException { + Connection conn = DriverManager.getConnection(jdbcURL, user, pwd); + conn.createStatement().execute("set hive.support.concurrency = false"); + return conn; + } + + @After + public void tearDown() throws Exception { + LlapBaseInputFormat.closeAll(); + hs2Conn.close(); + } + + + + @Test + public void testKillQueryById() throws Exception { + String user = System.getProperty("user.name"); + Connection con1 = getConnection(miniHS2.getJdbcURL(testDbName), + user, "bar"); + + final HiveStatement stmt = (HiveStatement)con1.createStatement(); + final StringBuffer stmtQueryId = new StringBuffer(); + ExceptionHolder originalQExHolder = new ExceptionHolder(); + originalQExHolder.throwable = null; + + // Thread executing the query + Thread tExecute = new Thread(new Runnable() { + @Override + public void run() { + try { + System.out.println("Executing query: "); + stmt.execute("set hive.llap.execution.mode = none"); + + // The test table has 500 rows, so total query time should be ~ 500*500ms + stmt.executeAsync("select sleepMsUDF(t1.int_col, 100), t1.int_col, t2.int_col " + + "from " + tableName + " t1 join " + tableName + " t2 on t1.int_col = t2.int_col"); + stmtQueryId.append(stmt.getQueryId()); + stmt.getUpdateCount(); + } catch (SQLException e) { + originalQExHolder.throwable = e; + } + } + }); + + tExecute.start(); + + // wait for other thread to create the stmt handle + int count = 0; + while (++count <= 10) { + Thread.sleep(2000); + String queryId; + if (stmtQueryId.length() != 0) { + queryId = stmtQueryId.toString(); + } else { + continue; Review comment: Shall remove else part if we move rest of the code with break inside "if" block. ########## File path: common/src/java/org/apache/hadoop/hive/conf/HiveConf.java ########## @@ -4979,10 +4979,11 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal HIVE_QUERY_REEXECUTION_ENABLED("hive.query.reexecution.enabled", true, "Enable query reexecutions"), - HIVE_QUERY_REEXECUTION_STRATEGIES("hive.query.reexecution.strategies", "overlay,reoptimize", + HIVE_QUERY_REEXECUTION_STRATEGIES("hive.query.reexecution.strategies", "overlay,reoptimize,reexecute_lost_am", "comma separated list of plugin can be used:\n" + " overlay: hiveconf subtree 'reexec.overlay' is used as an overlay in case of an execution errors out\n" - + " reoptimize: collects operator statistics during execution and recompile the query after a failure"), + + " reoptimize: collects operator statistics during execution and recompile the query after a failure\n" + + " reexecute_lost_am: reexecutes query if it failed due to tez am node gets decommissioned"), Review comment: Not just decommissioned case. We retry, even if AM is killed manually. So, let's keep it "...due to tez AM being killed." ########## File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/reexec/TestReExecuteKilledTezAMQueryPlugin.java ########## @@ -0,0 +1,207 @@ +/* + * 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.hadoop.hive.ql.reexec; + +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.LlapBaseInputFormat; +import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.yarn.api.records.ApplicationReport; +import org.apache.hadoop.yarn.api.records.YarnApplicationState; +import org.apache.hadoop.yarn.client.api.YarnClient; +import org.apache.hive.jdbc.BaseJdbcWithMiniLlap; Review comment: Can see unused imports. Pls remove them. ########## File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/reexec/TestReExecuteKilledTezAMQueryPlugin.java ########## @@ -0,0 +1,207 @@ +/* + * 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.hadoop.hive.ql.reexec; + +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.LlapBaseInputFormat; +import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.yarn.api.records.ApplicationReport; +import org.apache.hadoop.yarn.api.records.YarnApplicationState; +import org.apache.hadoop.yarn.client.api.YarnClient; +import org.apache.hive.jdbc.BaseJdbcWithMiniLlap; +import org.apache.hive.jdbc.HiveStatement; +import org.apache.hive.jdbc.TestJdbcWithMiniLlapArrow; +import org.apache.hive.jdbc.miniHS2.MiniHS2; +import org.junit.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.net.URL; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + + +public class TestReExecuteKilledTezAMQueryPlugin { + protected static final Logger LOG = LoggerFactory.getLogger(TestJdbcWithMiniLlapArrow.class); + + private static MiniHS2 miniHS2 = null; + private static final String tableName = "testKillTezAmTbl"; + private static String dataFileDir; + private static final String testDbName = "testKillTezAmDb"; + protected static Connection hs2Conn = null; + private static HiveConf conf; + + private static class ExceptionHolder { + Throwable throwable; + } + + static HiveConf defaultConf() throws Exception { + String confDir = "../../data/conf/llap/"; + if (confDir != null && !confDir.isEmpty()) { + HiveConf.setHiveSiteLocation(new URL("file://"+ new File(confDir).toURI().getPath() + "/hive-site.xml")); + System.out.println("Setting hive-site: " + HiveConf.getHiveSiteLocation()); + } + HiveConf defaultConf = new HiveConf(); + defaultConf.setBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY, false); + defaultConf.setBoolVar(HiveConf.ConfVars.HIVE_SERVER2_ENABLE_DOAS, false); + defaultConf.addResource(new URL("file://" + new File(confDir).toURI().getPath() + "/tez-site.xml")); + return defaultConf; + } + + @BeforeClass + public static void beforeTest() throws Exception { + conf = defaultConf(); + conf.setVar(HiveConf.ConfVars.USERS_IN_ADMIN_ROLE, System.getProperty("user.name")); + conf.set(HiveConf.ConfVars.HIVE_QUERY_REEXECUTION_STRATEGIES.varname, "reexecute_lost_am"); + MiniHS2.cleanupLocalDir(); + Class.forName(MiniHS2.getJdbcDriverName()); + miniHS2 = new MiniHS2(conf, MiniHS2.MiniClusterType.LLAP); + dataFileDir = conf.get("test.data.files").replace('\\', '/').replace("c:", ""); + Map<String, String> confOverlay = new HashMap<String, String>(); + miniHS2.start(confOverlay); + miniHS2.getDFS().getFileSystem().mkdirs(new Path("/apps_staging_dir/anonymous")); + + Connection conDefault = getConnection(miniHS2.getJdbcURL(), + System.getProperty("user.name"), "bar"); + Statement stmt = conDefault.createStatement(); + String tblName = testDbName + "." + tableName; + Path dataFilePath = new Path(dataFileDir, "kv1.txt"); + String udfName = TestJdbcWithMiniLlapArrow.SleepMsUDF.class.getName(); + stmt.execute("drop database if exists " + testDbName + " cascade"); + stmt.execute("create database " + testDbName); + stmt.execute("set role admin"); + stmt.execute("dfs -put " + dataFilePath.toString() + " " + "kv1.txt"); + stmt.execute("use " + testDbName); + stmt.execute("create table " + tblName + " (int_col int, value string) "); + stmt.execute("load data inpath 'kv1.txt' into table " + tblName); + stmt.execute("create function sleepMsUDF as '" + udfName + "'"); + stmt.execute("grant select on table " + tblName + " to role public"); + + stmt.close(); + conDefault.close(); + } + + @AfterClass + public static void afterTest() { + if (miniHS2 != null && miniHS2.isStarted()) { + miniHS2.stop(); + } + } + + @Before + public void setUp() throws Exception { + hs2Conn = getConnection(miniHS2.getJdbcURL(), System.getProperty("user.name"), "bar"); + } + + public static Connection getConnection(String jdbcURL, String user, String pwd) throws SQLException { + Connection conn = DriverManager.getConnection(jdbcURL, user, pwd); + conn.createStatement().execute("set hive.support.concurrency = false"); + return conn; + } + + @After + public void tearDown() throws Exception { + LlapBaseInputFormat.closeAll(); + hs2Conn.close(); + } + + + + @Test + public void testKillQueryById() throws Exception { + String user = System.getProperty("user.name"); + Connection con1 = getConnection(miniHS2.getJdbcURL(testDbName), + user, "bar"); + + final HiveStatement stmt = (HiveStatement)con1.createStatement(); + final StringBuffer stmtQueryId = new StringBuffer(); + ExceptionHolder originalQExHolder = new ExceptionHolder(); + originalQExHolder.throwable = null; + + // Thread executing the query + Thread tExecute = new Thread(new Runnable() { + @Override + public void run() { + try { + System.out.println("Executing query: "); + stmt.execute("set hive.llap.execution.mode = none"); + + // The test table has 500 rows, so total query time should be ~ 500*500ms Review comment: Do you mean 500*100ms? ########## File path: ql/src/java/org/apache/hadoop/hive/ql/reexec/ReExecuteLostAMQueryPlugin.java ########## @@ -0,0 +1,75 @@ +/* + * 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.hadoop.hive.ql.reexec; + +import org.apache.hadoop.hive.ql.Driver; +import org.apache.hadoop.hive.ql.hooks.ExecuteWithHookContext; +import org.apache.hadoop.hive.ql.hooks.HookContext; +import org.apache.hadoop.hive.ql.plan.mapper.PlanMapper; + +import java.util.regex.Pattern; + +public class ReExecuteLostAMQueryPlugin implements IReExecutionPlugin { + private boolean retryPossible; + + // Lost am container have exit code -100, due to node failures. + private Pattern lostAMContainerErrorPattern = Pattern.compile(".*AM Container for .* exited .* exitCode: -100.*"); Review comment: Based on -100 exitcode, can we differentiate if the query was killed by "KILL QUERY" command or just Tez AM got killed? If killed by KILL QUERY, then we shouldn't retry. ########## File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/reexec/TestReExecuteKilledTezAMQueryPlugin.java ########## @@ -0,0 +1,207 @@ +/* + * 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.hadoop.hive.ql.reexec; + +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.LlapBaseInputFormat; +import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.yarn.api.records.ApplicationReport; +import org.apache.hadoop.yarn.api.records.YarnApplicationState; +import org.apache.hadoop.yarn.client.api.YarnClient; +import org.apache.hive.jdbc.BaseJdbcWithMiniLlap; +import org.apache.hive.jdbc.HiveStatement; +import org.apache.hive.jdbc.TestJdbcWithMiniLlapArrow; +import org.apache.hive.jdbc.miniHS2.MiniHS2; +import org.junit.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.net.URL; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + + +public class TestReExecuteKilledTezAMQueryPlugin { + protected static final Logger LOG = LoggerFactory.getLogger(TestJdbcWithMiniLlapArrow.class); Review comment: TestReExecuteKilledTezAMQueryPlugin.class can be used as logger argument. ########## File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/reexec/TestReExecuteKilledTezAMQueryPlugin.java ########## @@ -0,0 +1,207 @@ +/* + * 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.hadoop.hive.ql.reexec; + +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.LlapBaseInputFormat; +import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.yarn.api.records.ApplicationReport; +import org.apache.hadoop.yarn.api.records.YarnApplicationState; +import org.apache.hadoop.yarn.client.api.YarnClient; +import org.apache.hive.jdbc.BaseJdbcWithMiniLlap; +import org.apache.hive.jdbc.HiveStatement; +import org.apache.hive.jdbc.TestJdbcWithMiniLlapArrow; +import org.apache.hive.jdbc.miniHS2.MiniHS2; +import org.junit.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.net.URL; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + + +public class TestReExecuteKilledTezAMQueryPlugin { + protected static final Logger LOG = LoggerFactory.getLogger(TestJdbcWithMiniLlapArrow.class); + + private static MiniHS2 miniHS2 = null; + private static final String tableName = "testKillTezAmTbl"; + private static String dataFileDir; + private static final String testDbName = "testKillTezAmDb"; + protected static Connection hs2Conn = null; + private static HiveConf conf; + + private static class ExceptionHolder { + Throwable throwable; + } + + static HiveConf defaultConf() throws Exception { + String confDir = "../../data/conf/llap/"; + if (confDir != null && !confDir.isEmpty()) { + HiveConf.setHiveSiteLocation(new URL("file://"+ new File(confDir).toURI().getPath() + "/hive-site.xml")); + System.out.println("Setting hive-site: " + HiveConf.getHiveSiteLocation()); + } + HiveConf defaultConf = new HiveConf(); + defaultConf.setBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY, false); + defaultConf.setBoolVar(HiveConf.ConfVars.HIVE_SERVER2_ENABLE_DOAS, false); + defaultConf.addResource(new URL("file://" + new File(confDir).toURI().getPath() + "/tez-site.xml")); + return defaultConf; + } + + @BeforeClass + public static void beforeTest() throws Exception { + conf = defaultConf(); + conf.setVar(HiveConf.ConfVars.USERS_IN_ADMIN_ROLE, System.getProperty("user.name")); + conf.set(HiveConf.ConfVars.HIVE_QUERY_REEXECUTION_STRATEGIES.varname, "reexecute_lost_am"); + MiniHS2.cleanupLocalDir(); + Class.forName(MiniHS2.getJdbcDriverName()); + miniHS2 = new MiniHS2(conf, MiniHS2.MiniClusterType.LLAP); + dataFileDir = conf.get("test.data.files").replace('\\', '/').replace("c:", ""); + Map<String, String> confOverlay = new HashMap<String, String>(); + miniHS2.start(confOverlay); + miniHS2.getDFS().getFileSystem().mkdirs(new Path("/apps_staging_dir/anonymous")); + + Connection conDefault = getConnection(miniHS2.getJdbcURL(), + System.getProperty("user.name"), "bar"); + Statement stmt = conDefault.createStatement(); + String tblName = testDbName + "." + tableName; + Path dataFilePath = new Path(dataFileDir, "kv1.txt"); + String udfName = TestJdbcWithMiniLlapArrow.SleepMsUDF.class.getName(); + stmt.execute("drop database if exists " + testDbName + " cascade"); + stmt.execute("create database " + testDbName); + stmt.execute("set role admin"); + stmt.execute("dfs -put " + dataFilePath.toString() + " " + "kv1.txt"); + stmt.execute("use " + testDbName); + stmt.execute("create table " + tblName + " (int_col int, value string) "); + stmt.execute("load data inpath 'kv1.txt' into table " + tblName); + stmt.execute("create function sleepMsUDF as '" + udfName + "'"); + stmt.execute("grant select on table " + tblName + " to role public"); + + stmt.close(); + conDefault.close(); + } + + @AfterClass + public static void afterTest() { + if (miniHS2 != null && miniHS2.isStarted()) { + miniHS2.stop(); + } + } + + @Before + public void setUp() throws Exception { + hs2Conn = getConnection(miniHS2.getJdbcURL(), System.getProperty("user.name"), "bar"); + } + + public static Connection getConnection(String jdbcURL, String user, String pwd) throws SQLException { + Connection conn = DriverManager.getConnection(jdbcURL, user, pwd); + conn.createStatement().execute("set hive.support.concurrency = false"); + return conn; + } + + @After + public void tearDown() throws Exception { + LlapBaseInputFormat.closeAll(); + hs2Conn.close(); + } + + + + @Test + public void testKillQueryById() throws Exception { + String user = System.getProperty("user.name"); + Connection con1 = getConnection(miniHS2.getJdbcURL(testDbName), + user, "bar"); + + final HiveStatement stmt = (HiveStatement)con1.createStatement(); + final StringBuffer stmtQueryId = new StringBuffer(); + ExceptionHolder originalQExHolder = new ExceptionHolder(); + originalQExHolder.throwable = null; + + // Thread executing the query + Thread tExecute = new Thread(new Runnable() { + @Override + public void run() { + try { + System.out.println("Executing query: "); + stmt.execute("set hive.llap.execution.mode = none"); + + // The test table has 500 rows, so total query time should be ~ 500*500ms + stmt.executeAsync("select sleepMsUDF(t1.int_col, 100), t1.int_col, t2.int_col " + + "from " + tableName + " t1 join " + tableName + " t2 on t1.int_col = t2.int_col"); + stmtQueryId.append(stmt.getQueryId()); + stmt.getUpdateCount(); + } catch (SQLException e) { + originalQExHolder.throwable = e; + } + } + }); + + tExecute.start(); + + // wait for other thread to create the stmt handle + int count = 0; + while (++count <= 10) { + Thread.sleep(2000); + String queryId; + if (stmtQueryId.length() != 0) { + queryId = stmtQueryId.toString(); + } else { + continue; + } + System.out.println("Killing query: " + queryId); + killAMForQueryId(queryId); + break; + } + + tExecute.join(); + Assert.assertEquals(originalQExHolder.throwable, null); + try { Review comment: We shall move cleanup code before Assert as it may not execute if Assert throws exception. ########## File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/reexec/TestReExecuteKilledTezAMQueryPlugin.java ########## @@ -0,0 +1,207 @@ +/* + * 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.hadoop.hive.ql.reexec; + +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.llap.LlapBaseInputFormat; +import org.apache.hadoop.hive.ql.metadata.HiveException; +import org.apache.hadoop.yarn.api.records.ApplicationReport; +import org.apache.hadoop.yarn.api.records.YarnApplicationState; +import org.apache.hadoop.yarn.client.api.YarnClient; +import org.apache.hive.jdbc.BaseJdbcWithMiniLlap; +import org.apache.hive.jdbc.HiveStatement; +import org.apache.hive.jdbc.TestJdbcWithMiniLlapArrow; +import org.apache.hive.jdbc.miniHS2.MiniHS2; +import org.junit.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.net.URL; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + + +public class TestReExecuteKilledTezAMQueryPlugin { + protected static final Logger LOG = LoggerFactory.getLogger(TestJdbcWithMiniLlapArrow.class); + + private static MiniHS2 miniHS2 = null; + private static final String tableName = "testKillTezAmTbl"; + private static String dataFileDir; + private static final String testDbName = "testKillTezAmDb"; + protected static Connection hs2Conn = null; + private static HiveConf conf; + + private static class ExceptionHolder { + Throwable throwable; + } + + static HiveConf defaultConf() throws Exception { + String confDir = "../../data/conf/llap/"; + if (confDir != null && !confDir.isEmpty()) { + HiveConf.setHiveSiteLocation(new URL("file://"+ new File(confDir).toURI().getPath() + "/hive-site.xml")); + System.out.println("Setting hive-site: " + HiveConf.getHiveSiteLocation()); + } + HiveConf defaultConf = new HiveConf(); + defaultConf.setBoolVar(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY, false); + defaultConf.setBoolVar(HiveConf.ConfVars.HIVE_SERVER2_ENABLE_DOAS, false); + defaultConf.addResource(new URL("file://" + new File(confDir).toURI().getPath() + "/tez-site.xml")); + return defaultConf; + } + + @BeforeClass + public static void beforeTest() throws Exception { + conf = defaultConf(); + conf.setVar(HiveConf.ConfVars.USERS_IN_ADMIN_ROLE, System.getProperty("user.name")); + conf.set(HiveConf.ConfVars.HIVE_QUERY_REEXECUTION_STRATEGIES.varname, "reexecute_lost_am"); + MiniHS2.cleanupLocalDir(); + Class.forName(MiniHS2.getJdbcDriverName()); + miniHS2 = new MiniHS2(conf, MiniHS2.MiniClusterType.LLAP); + dataFileDir = conf.get("test.data.files").replace('\\', '/').replace("c:", ""); + Map<String, String> confOverlay = new HashMap<String, String>(); + miniHS2.start(confOverlay); + miniHS2.getDFS().getFileSystem().mkdirs(new Path("/apps_staging_dir/anonymous")); + + Connection conDefault = getConnection(miniHS2.getJdbcURL(), + System.getProperty("user.name"), "bar"); + Statement stmt = conDefault.createStatement(); + String tblName = testDbName + "." + tableName; + Path dataFilePath = new Path(dataFileDir, "kv1.txt"); + String udfName = TestJdbcWithMiniLlapArrow.SleepMsUDF.class.getName(); + stmt.execute("drop database if exists " + testDbName + " cascade"); + stmt.execute("create database " + testDbName); + stmt.execute("set role admin"); + stmt.execute("dfs -put " + dataFilePath.toString() + " " + "kv1.txt"); + stmt.execute("use " + testDbName); + stmt.execute("create table " + tblName + " (int_col int, value string) "); + stmt.execute("load data inpath 'kv1.txt' into table " + tblName); + stmt.execute("create function sleepMsUDF as '" + udfName + "'"); + stmt.execute("grant select on table " + tblName + " to role public"); + + stmt.close(); + conDefault.close(); + } + + @AfterClass + public static void afterTest() { + if (miniHS2 != null && miniHS2.isStarted()) { + miniHS2.stop(); + } + } + + @Before + public void setUp() throws Exception { + hs2Conn = getConnection(miniHS2.getJdbcURL(), System.getProperty("user.name"), "bar"); + } + + public static Connection getConnection(String jdbcURL, String user, String pwd) throws SQLException { + Connection conn = DriverManager.getConnection(jdbcURL, user, pwd); + conn.createStatement().execute("set hive.support.concurrency = false"); + return conn; + } + + @After + public void tearDown() throws Exception { + LlapBaseInputFormat.closeAll(); Review comment: Why need LlapBaseInputFormat.closeAll? ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
