abstractdog commented on code in PR #6640: URL: https://github.com/apache/hive/pull/6640#discussion_r3842133533
########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestHiveServer2Connectivity.java: ########## @@ -0,0 +1,129 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; + +public class TestHiveServer2Connectivity { + + private static TezYarnClusterContainer cluster; + private static HiveServer2 hs2; + private static int hs2Port; + private static String hdfsUri; + + @BeforeClass + public static void startAll() throws Exception { + cluster = new TezYarnClusterContainer(); + cluster.start(); + + hdfsUri = cluster.getHdfsUri(); + + cluster.namenodeContainer().execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp/hive-29483/warehouse"); + cluster.namenodeContainer().execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp/hive-29483/scratch"); + cluster.namenodeContainer().execInContainer("hdfs", "dfs", "-chmod", "-R", "777", "/tmp/hive-29483"); + + Path localScratch = Files.createTempDirectory("hive-29483-local-"); + + HiveConf conf = new HiveConf(); + URL hiveSite = TestHiveServer2Connectivity.class.getClassLoader().getResource("hive-site-yarn-it.xml"); + URL yarnSite = TestHiveServer2Connectivity.class.getClassLoader().getResource("yarn-site.xml"); + if (hiveSite != null) conf.addResource(hiveSite); + if (yarnSite != null) conf.addResource(yarnSite); Review Comment: use proper `{}` blocks for single line blocks as well ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestHiveServer2Connectivity.java: ########## @@ -0,0 +1,129 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; + +public class TestHiveServer2Connectivity { + + private static TezYarnClusterContainer cluster; + private static HiveServer2 hs2; + private static int hs2Port; + private static String hdfsUri; + + @BeforeClass + public static void startAll() throws Exception { + cluster = new TezYarnClusterContainer(); + cluster.start(); + + hdfsUri = cluster.getHdfsUri(); + + cluster.namenodeContainer().execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp/hive-29483/warehouse"); + cluster.namenodeContainer().execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp/hive-29483/scratch"); + cluster.namenodeContainer().execInContainer("hdfs", "dfs", "-chmod", "-R", "777", "/tmp/hive-29483"); + + Path localScratch = Files.createTempDirectory("hive-29483-local-"); Review Comment: using "29483" explicitly looks strange, it should be handled by the test code under the hood ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestTezYarnLocalization.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class TestTezYarnLocalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestTezYarnLocalization.class); + + private static final String HDFS_BASE = "hdfs://namenode:8020"; + private static final String HDFS_WAREHOUSE = HDFS_BASE + "/tmp/hive-tez-loc/warehouse"; + private static final String HDFS_SCRATCH = HDFS_BASE + "/tmp/hive-tez-loc/scratch"; + private static final String HDFS_ROOT = "/tmp/hive-tez-loc"; Review Comment: no need for extra indendation for aligning values ########## itests/tez-yarn-it/pom.xml: ########## @@ -0,0 +1,171 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.hive</groupId> + <artifactId>hive-it</artifactId> + <version>4.3.0-SNAPSHOT</version> + <relativePath>../pom.xml</relativePath> + </parent> + <artifactId>hive-it-tez-yarn</artifactId> + <packaging>jar</packaging> + <name>Hive Integration - Tez on YARN localization tests</name> + <properties> + <hive.path.to.root>../..</hive.path.to.root> + <skip.tez.yarn.tests>true</skip.tez.yarn.tests> + </properties> + <dependencies> + <dependency> + <groupId>org.testcontainers</groupId> + <artifactId>testcontainers</artifactId> + <scope>test</scope> + </dependency> + <!-- Explicit Tez dependencies: Hive's Tez compilation/execution path needs these classes + on the in-process HS2 JVM classpath. In the main Hive build they are often profile-gated. --> + <dependency> + <groupId>org.apache.tez</groupId> + <artifactId>tez-api</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.tez</groupId> + <artifactId>tez-common</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.tez</groupId> + <artifactId>tez-dag</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.tez</groupId> + <artifactId>tez-runtime-library</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.tez</groupId> + <artifactId>tez-runtime-internals</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.tez</groupId> + <artifactId>tez-mapreduce</artifactId> + <scope>test</scope> + </dependency> + <!-- Provides org.apache.tez.hadoop.shim.HadoopShimsLoader, which DAGAppMaster.serviceInit() + loads at Tez AM startup. Normally transitive via tez-api; declared explicitly so it is + reliably on the classpath and gets staged into tez.lib.uris (its jar name has no "tez"). --> + <dependency> + <groupId>org.apache.tez</groupId> + <artifactId>hadoop-shim</artifactId> + <version>${tez.version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.hive</groupId> + <artifactId>hive-service</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.hive</groupId> + <artifactId>hive-jdbc</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.hive</groupId> + <artifactId>hive-exec</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.hive</groupId> + <artifactId>hive-it-util</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-common</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-mapreduce-client-core</artifactId> + <scope>test</scope> + </dependency> + <dependency> + <groupId>junit</groupId> + <artifactId>junit</artifactId> + <scope>test</scope> + </dependency> + </dependencies> + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-antrun-plugin</artifactId> + <executions> + <execution> + <id>generate-custom-hosts-file</id> + <!-- Must run after the parent "setup-test-dirs" execution (process-test-resources). --> + <phase>test-compile</phase> + <goals> + <goal>run</goal> + </goals> + <configuration> + <target> + <mkdir dir="${test.conf.dir}"/> + <copy file="${basedir}/src/test/resources/custom_hosts_file" tofile="${test.conf.dir}/custom_hosts_file" overwrite="true"/> + <exec executable="hostname" output="${test.conf.dir}/local_hostname.txt" failonerror="true"/> + <loadfile property="hive.tez.yarn.it.local.hostname" srcFile="${test.conf.dir}/local_hostname.txt"> + <filterchain> + <striplinebreaks/> + </filterchain> + </loadfile> + <!-- + When -Djdk.net.hosts.file is set, Java hostname resolution uses only that file. + Ensure the local machine hostname is resolvable so Hive can start/compile queries. + --> + <echo file="${test.conf.dir}/custom_hosts_file" append="true">127.0.0.1 ${hive.tez.yarn.it.local.hostname}</echo> + </target> + </configuration> + </execution> + </executions> + </plugin> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-surefire-plugin</artifactId> + <configuration> + <skipTests>${skip.tez.yarn.tests}</skipTests> + <!-- Resolve docker-network hostnames (namenode, resourcemanager, …) to 127.0.0.1 + so HDFS URIs embedded in Tez LocalResources are reachable from both the host JVM + and inside YARN containers (via the shared docker network alias). --> + <argLine>${maven.test.jvm.args} -Xshare:off -Djdk.net.hosts.file=${test.conf.dir}/custom_hosts_file</argLine> Review Comment: is there a specific reason to disable class data sharing via Xshare? if the base java image supports class metadata sharing for the base JDK classes, it can reduce container startup time a bit ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestTezYarnLocalization.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class TestTezYarnLocalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestTezYarnLocalization.class); + + private static final String HDFS_BASE = "hdfs://namenode:8020"; + private static final String HDFS_WAREHOUSE = HDFS_BASE + "/tmp/hive-tez-loc/warehouse"; + private static final String HDFS_SCRATCH = HDFS_BASE + "/tmp/hive-tez-loc/scratch"; + private static final String HDFS_ROOT = "/tmp/hive-tez-loc"; + + private static TezYarnClusterContainer cluster; + private static HiveServer2 hs2; + private static int hs2Port; + + @BeforeClass + public static void startAll() throws Exception { + cluster = new TezYarnClusterContainer(true); + cluster.start(); + + GenericContainer<?> nn = cluster.namenodeContainer(); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", "/tmp"); + + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/warehouse"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/scratch"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/user-install"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", HDFS_ROOT); + + String tezLibUris = cluster.uploadTezLibsToHdfs(); + LOG.info("Staged Tez libs to HDFS: {}", tezLibUris); + + Path localScratch = Files.createDirectories( + Path.of("/tmp", "hive-tez-loc-" + System.currentTimeMillis())); + HiveConf conf = buildHiveConf(tezLibUris, localScratch); + + hs2 = new HiveServer2(); + hs2.init(conf); + hs2.start(); + + waitForJdbc(hs2Port); + LOG.info("HiveServer2 is ready on port {}", hs2Port); + } + + @AfterClass + public static void stopAll() { + dumpNodeManagerDiagnostics(); + if (hs2 != null) { + hs2.stop(); + hs2 = null; + } + if (cluster != null) { + cluster.stop(); + cluster = null; + } + } + + @Test + public void testQuerySucceedsWithAppJar() throws Exception { + String url = jdbcUrl(hs2Port); + try (Connection conn = DriverManager.getConnection(url, "hive", "")) { + try (Statement stmt = conn.createStatement()) { + + stmt.execute("CREATE TABLE IF NOT EXISTS tez_loc_test (id INT) STORED AS ORC"); + stmt.execute("CREATE TABLE IF NOT EXISTS tez_source (id INT) STORED AS ORC"); + + stmt.execute("INSERT INTO tez_loc_test SELECT count(*) FROM tez_source"); + + try (ResultSet rs = stmt.executeQuery("SELECT id FROM tez_loc_test")) { + Assert.assertTrue("Result set must contain at least one row", rs.next()); + long count = rs.getLong(1); + Assert.assertEquals( + "INSERT SELECT count(*) FROM empty tez_source should return 0 (hive-exec.jar was localized)", + 0L, count); + LOG.info("Tez query succeeded: inserted count(*) = {}", count); + } + } + } + + verifyTezYarnAppExists(); + } + + /** Prints Tez AM and NodeManager logs to the Surefire *-output.txt file at teardown. */ + private static void dumpNodeManagerDiagnostics() { + if (cluster == null) { + return; + } + System.out.println("########## BEGIN NodeManager diagnostics ##########"); Review Comment: use logging library ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestTezYarnLocalization.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class TestTezYarnLocalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestTezYarnLocalization.class); + + private static final String HDFS_BASE = "hdfs://namenode:8020"; + private static final String HDFS_WAREHOUSE = HDFS_BASE + "/tmp/hive-tez-loc/warehouse"; + private static final String HDFS_SCRATCH = HDFS_BASE + "/tmp/hive-tez-loc/scratch"; + private static final String HDFS_ROOT = "/tmp/hive-tez-loc"; + + private static TezYarnClusterContainer cluster; + private static HiveServer2 hs2; + private static int hs2Port; + + @BeforeClass + public static void startAll() throws Exception { + cluster = new TezYarnClusterContainer(true); + cluster.start(); + + GenericContainer<?> nn = cluster.namenodeContainer(); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", "/tmp"); + + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/warehouse"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/scratch"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/user-install"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", HDFS_ROOT); + + String tezLibUris = cluster.uploadTezLibsToHdfs(); + LOG.info("Staged Tez libs to HDFS: {}", tezLibUris); + + Path localScratch = Files.createDirectories( + Path.of("/tmp", "hive-tez-loc-" + System.currentTimeMillis())); + HiveConf conf = buildHiveConf(tezLibUris, localScratch); + + hs2 = new HiveServer2(); + hs2.init(conf); + hs2.start(); + + waitForJdbc(hs2Port); + LOG.info("HiveServer2 is ready on port {}", hs2Port); + } + + @AfterClass + public static void stopAll() { + dumpNodeManagerDiagnostics(); + if (hs2 != null) { + hs2.stop(); + hs2 = null; + } + if (cluster != null) { + cluster.stop(); + cluster = null; + } + } + + @Test + public void testQuerySucceedsWithAppJar() throws Exception { + String url = jdbcUrl(hs2Port); + try (Connection conn = DriverManager.getConnection(url, "hive", "")) { + try (Statement stmt = conn.createStatement()) { + + stmt.execute("CREATE TABLE IF NOT EXISTS tez_loc_test (id INT) STORED AS ORC"); + stmt.execute("CREATE TABLE IF NOT EXISTS tez_source (id INT) STORED AS ORC"); + + stmt.execute("INSERT INTO tez_loc_test SELECT count(*) FROM tez_source"); + + try (ResultSet rs = stmt.executeQuery("SELECT id FROM tez_loc_test")) { + Assert.assertTrue("Result set must contain at least one row", rs.next()); + long count = rs.getLong(1); + Assert.assertEquals( + "INSERT SELECT count(*) FROM empty tez_source should return 0 (hive-exec.jar was localized)", + 0L, count); + LOG.info("Tez query succeeded: inserted count(*) = {}", count); + } + } + } + + verifyTezYarnAppExists(); + } + + /** Prints Tez AM and NodeManager logs to the Surefire *-output.txt file at teardown. */ + private static void dumpNodeManagerDiagnostics() { + if (cluster == null) { + return; + } + System.out.println("########## BEGIN NodeManager diagnostics ##########"); + try { + dumpNmCommand("launch_container.sh (AM launch command + classpath)", + "find /tmp -name 'launch_container.sh' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container syslog (Tez AM log4j output)", + "find /var/log/hadoop/userlogs -name 'syslog*' 2>/dev/null | head -10 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container stdout + stderr + prelaunch.err", + "find /var/log/hadoop/userlogs \\( -name 'stdout' -o -name 'stderr' -o -name 'prelaunch.err' \\) " + + "2>/dev/null | head -20 | xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("NodeManager daemon log (tail 200)", + "find /var/log/hadoop -maxdepth 1 -name '*.log' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; tail -200 {}' 2>/dev/null || true"); + } catch (Exception e) { + System.out.println("Could not dump NodeManager diagnostics: " + e); + } + System.out.println("########## END NodeManager diagnostics ##########"); + System.out.flush(); + } + + private static void dumpNmCommand(String label, String bashCommand) { + try { + GenericContainer.ExecResult r = + cluster.nodeManagerContainer().execInContainer("bash", "-c", bashCommand); + String out = r.getStdout(); + System.out.println("===== NM: " + label + " ====="); + System.out.println(out.isEmpty() ? "(no output found)" : out); + } catch (Exception e) { + System.out.println("===== NM: " + label + " (dump failed: " + e + ") ====="); + } + } + + private static HiveConf buildHiveConf(String tezLibUris, Path localScratch) throws Exception { + HiveConf conf = new HiveConf(); + + URL hiveSite = TestTezYarnLocalization.class.getClassLoader().getResource("hive-site-yarn-it.xml"); + URL yarnSite = TestTezYarnLocalization.class.getClassLoader().getResource("yarn-site.xml"); + if (hiveSite != null) { conf.addResource(hiveSite); } + if (yarnSite != null) { conf.addResource(yarnSite); } + + conf.set("fs.defaultFS", HDFS_BASE); + conf.setBoolean("dfs.client.use.datanode.hostname", true); + conf.set("hive.metastore.warehouse.dir", HDFS_WAREHOUSE); + conf.set(HiveConf.ConfVars.SCRATCH_DIR.varname, HDFS_SCRATCH); + conf.set(HiveConf.ConfVars.LOCAL_SCRATCH_DIR.varname, localScratch.toAbsolutePath().toString()); + conf.setVar(HiveConf.ConfVars.HIVE_USER_INSTALL_DIR, HDFS_ROOT + "/user-install"); + + conf.set("javax.jdo.option.ConnectionURL", + "jdbc:derby:" + localScratch.resolve("metastore_db").toAbsolutePath() + ";create=true"); + + conf.setBoolVar(HiveConf.ConfVars.METASTORE_TRY_DIRECT_SQL, false); + conf.set("hive.stats.autogather", "false"); + conf.set("hive.stats.column.autogather", "false"); + conf.set("yarn.resourcemanager.hostname", "resourcemanager"); + conf.set("yarn.resourcemanager.address", "resourcemanager:8032"); + conf.set("yarn.resourcemanager.webapp.address", "resourcemanager:8088"); Review Comment: no need for extra indentation for aligment ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TezYarnClusterContainer.java: ########## @@ -0,0 +1,377 @@ +/* + * 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.hive.tez.yarn; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.FixedHostPortGenericContainer; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.images.builder.ImageFromDockerfile; +import org.testcontainers.utility.MountableFile; + +import java.io.File; +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.security.CodeSource; +import java.time.Duration; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Stream; + +public class TezYarnClusterContainer { + + private static final Logger LOG = LoggerFactory.getLogger(TezYarnClusterContainer.class); + + /** Path to the Java 21 runtime inside containers for Tez AM/task launch environments. */ + public static final String CONTAINER_JAVA_21_HOME = "/opt/jdk21"; + + private static final String HADOOP_IMAGE = buildHadoopImage(); + private static final Duration STARTUP_TIMEOUT = Duration.ofMinutes(3); + private static final Map<String, String> COMMON_ENV = loadCommonEnv(); + + private static final int NN_RPC_PORT = 8020; + private static final int NN_HTTP_PORT = 9870; + private static final int RM_RPC_PORT = 8032; + private static final int RM_HTTP_PORT = 8088; + private static final int DN_HTTP_PORT = 9864; + private static final int DN_XFER_PORT = 9866; + // Tez AM client RPC port, published by the NM container so the host JVM can reach the AM. + public static final int AM_CLIENT_PORT = 41000; + + private final Network network; + private final GenericContainer<?> namenode; + private final GenericContainer<?> datanode; + private final GenericContainer<?> resourcemanager; + private final GenericContainer<?> nodemanager; + private final boolean fixedPorts; + + public TezYarnClusterContainer() { + this(false); + } + + public TezYarnClusterContainer(boolean fixedPorts) { + this.fixedPorts = fixedPorts; + network = Network.newNetwork(); + + if (fixedPorts) { Review Comment: this part is a bit verbose, I would eliminate the duplicated parts which are the same regardless of `fixedPorts`: ``` .withNetwork(network) .withNetworkAliases("namenode") .withCommand("hdfs", "namenode") .withEnv(COMMON_ENV) .withEnv("ENSURE_NAMENODE_DIR", "/tmp/hadoop-hadoop/dfs/name") .waitingFor(Wait.forHttp("/").forPort(NN_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); ``` ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestTezYarnLocalization.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class TestTezYarnLocalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestTezYarnLocalization.class); + + private static final String HDFS_BASE = "hdfs://namenode:8020"; + private static final String HDFS_WAREHOUSE = HDFS_BASE + "/tmp/hive-tez-loc/warehouse"; + private static final String HDFS_SCRATCH = HDFS_BASE + "/tmp/hive-tez-loc/scratch"; + private static final String HDFS_ROOT = "/tmp/hive-tez-loc"; + + private static TezYarnClusterContainer cluster; + private static HiveServer2 hs2; + private static int hs2Port; + + @BeforeClass + public static void startAll() throws Exception { + cluster = new TezYarnClusterContainer(true); + cluster.start(); + + GenericContainer<?> nn = cluster.namenodeContainer(); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", "/tmp"); + + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/warehouse"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/scratch"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/user-install"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", HDFS_ROOT); + + String tezLibUris = cluster.uploadTezLibsToHdfs(); + LOG.info("Staged Tez libs to HDFS: {}", tezLibUris); + + Path localScratch = Files.createDirectories( + Path.of("/tmp", "hive-tez-loc-" + System.currentTimeMillis())); + HiveConf conf = buildHiveConf(tezLibUris, localScratch); + + hs2 = new HiveServer2(); + hs2.init(conf); + hs2.start(); + + waitForJdbc(hs2Port); + LOG.info("HiveServer2 is ready on port {}", hs2Port); + } + + @AfterClass + public static void stopAll() { + dumpNodeManagerDiagnostics(); + if (hs2 != null) { + hs2.stop(); + hs2 = null; + } + if (cluster != null) { + cluster.stop(); + cluster = null; + } + } + + @Test + public void testQuerySucceedsWithAppJar() throws Exception { + String url = jdbcUrl(hs2Port); + try (Connection conn = DriverManager.getConnection(url, "hive", "")) { + try (Statement stmt = conn.createStatement()) { + + stmt.execute("CREATE TABLE IF NOT EXISTS tez_loc_test (id INT) STORED AS ORC"); + stmt.execute("CREATE TABLE IF NOT EXISTS tez_source (id INT) STORED AS ORC"); + + stmt.execute("INSERT INTO tez_loc_test SELECT count(*) FROM tez_source"); + + try (ResultSet rs = stmt.executeQuery("SELECT id FROM tez_loc_test")) { + Assert.assertTrue("Result set must contain at least one row", rs.next()); + long count = rs.getLong(1); + Assert.assertEquals( + "INSERT SELECT count(*) FROM empty tez_source should return 0 (hive-exec.jar was localized)", + 0L, count); + LOG.info("Tez query succeeded: inserted count(*) = {}", count); + } + } + } + + verifyTezYarnAppExists(); + } + + /** Prints Tez AM and NodeManager logs to the Surefire *-output.txt file at teardown. */ + private static void dumpNodeManagerDiagnostics() { + if (cluster == null) { + return; + } + System.out.println("########## BEGIN NodeManager diagnostics ##########"); + try { + dumpNmCommand("launch_container.sh (AM launch command + classpath)", + "find /tmp -name 'launch_container.sh' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container syslog (Tez AM log4j output)", + "find /var/log/hadoop/userlogs -name 'syslog*' 2>/dev/null | head -10 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container stdout + stderr + prelaunch.err", + "find /var/log/hadoop/userlogs \\( -name 'stdout' -o -name 'stderr' -o -name 'prelaunch.err' \\) " + + "2>/dev/null | head -20 | xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("NodeManager daemon log (tail 200)", + "find /var/log/hadoop -maxdepth 1 -name '*.log' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; tail -200 {}' 2>/dev/null || true"); + } catch (Exception e) { + System.out.println("Could not dump NodeManager diagnostics: " + e); + } + System.out.println("########## END NodeManager diagnostics ##########"); + System.out.flush(); + } + + private static void dumpNmCommand(String label, String bashCommand) { + try { + GenericContainer.ExecResult r = + cluster.nodeManagerContainer().execInContainer("bash", "-c", bashCommand); + String out = r.getStdout(); + System.out.println("===== NM: " + label + " ====="); + System.out.println(out.isEmpty() ? "(no output found)" : out); + } catch (Exception e) { + System.out.println("===== NM: " + label + " (dump failed: " + e + ") ====="); + } + } + + private static HiveConf buildHiveConf(String tezLibUris, Path localScratch) throws Exception { + HiveConf conf = new HiveConf(); + + URL hiveSite = TestTezYarnLocalization.class.getClassLoader().getResource("hive-site-yarn-it.xml"); + URL yarnSite = TestTezYarnLocalization.class.getClassLoader().getResource("yarn-site.xml"); + if (hiveSite != null) { conf.addResource(hiveSite); } + if (yarnSite != null) { conf.addResource(yarnSite); } Review Comment: line break for blocks ########## itests/tez-yarn-it/pom.xml: ########## @@ -0,0 +1,171 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + 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. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.hive</groupId> + <artifactId>hive-it</artifactId> + <version>4.3.0-SNAPSHOT</version> + <relativePath>../pom.xml</relativePath> + </parent> + <artifactId>hive-it-tez-yarn</artifactId> + <packaging>jar</packaging> + <name>Hive Integration - Tez on YARN localization tests</name> + <properties> + <hive.path.to.root>../..</hive.path.to.root> + <skip.tez.yarn.tests>true</skip.tez.yarn.tests> Review Comment: instead of `skip.tez.yarn.tests` I would use `run.tez.yarn.tests`, because `skip` + `false` is always a strange form of double-negative ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestTezYarnLocalization.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class TestTezYarnLocalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestTezYarnLocalization.class); + + private static final String HDFS_BASE = "hdfs://namenode:8020"; + private static final String HDFS_WAREHOUSE = HDFS_BASE + "/tmp/hive-tez-loc/warehouse"; + private static final String HDFS_SCRATCH = HDFS_BASE + "/tmp/hive-tez-loc/scratch"; + private static final String HDFS_ROOT = "/tmp/hive-tez-loc"; + + private static TezYarnClusterContainer cluster; + private static HiveServer2 hs2; + private static int hs2Port; + + @BeforeClass + public static void startAll() throws Exception { + cluster = new TezYarnClusterContainer(true); + cluster.start(); + + GenericContainer<?> nn = cluster.namenodeContainer(); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", "/tmp"); + + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/warehouse"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/scratch"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/user-install"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", HDFS_ROOT); + + String tezLibUris = cluster.uploadTezLibsToHdfs(); + LOG.info("Staged Tez libs to HDFS: {}", tezLibUris); + + Path localScratch = Files.createDirectories( + Path.of("/tmp", "hive-tez-loc-" + System.currentTimeMillis())); + HiveConf conf = buildHiveConf(tezLibUris, localScratch); + + hs2 = new HiveServer2(); + hs2.init(conf); + hs2.start(); + + waitForJdbc(hs2Port); + LOG.info("HiveServer2 is ready on port {}", hs2Port); + } + + @AfterClass + public static void stopAll() { + dumpNodeManagerDiagnostics(); + if (hs2 != null) { + hs2.stop(); + hs2 = null; + } + if (cluster != null) { + cluster.stop(); + cluster = null; + } + } + + @Test + public void testQuerySucceedsWithAppJar() throws Exception { Review Comment: this unit test might want to become even smarter by asserting 2 more things: 1. hive-exec.jar is present on HDFS somewhere (result of localization step 1) 2. hive-exec.jar is present in the yarn container's local folder (result of localization step 2) ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TezYarnClusterContainer.java: ########## @@ -0,0 +1,377 @@ +/* + * 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.hive.tez.yarn; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.FixedHostPortGenericContainer; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.images.builder.ImageFromDockerfile; +import org.testcontainers.utility.MountableFile; + +import java.io.File; +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.security.CodeSource; +import java.time.Duration; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Stream; + +public class TezYarnClusterContainer { + + private static final Logger LOG = LoggerFactory.getLogger(TezYarnClusterContainer.class); + + /** Path to the Java 21 runtime inside containers for Tez AM/task launch environments. */ + public static final String CONTAINER_JAVA_21_HOME = "/opt/jdk21"; + + private static final String HADOOP_IMAGE = buildHadoopImage(); + private static final Duration STARTUP_TIMEOUT = Duration.ofMinutes(3); + private static final Map<String, String> COMMON_ENV = loadCommonEnv(); + + private static final int NN_RPC_PORT = 8020; + private static final int NN_HTTP_PORT = 9870; + private static final int RM_RPC_PORT = 8032; + private static final int RM_HTTP_PORT = 8088; + private static final int DN_HTTP_PORT = 9864; + private static final int DN_XFER_PORT = 9866; + // Tez AM client RPC port, published by the NM container so the host JVM can reach the AM. + public static final int AM_CLIENT_PORT = 41000; + + private final Network network; + private final GenericContainer<?> namenode; + private final GenericContainer<?> datanode; + private final GenericContainer<?> resourcemanager; + private final GenericContainer<?> nodemanager; + private final boolean fixedPorts; + + public TezYarnClusterContainer() { + this(false); + } + + public TezYarnClusterContainer(boolean fixedPorts) { + this.fixedPorts = fixedPorts; + network = Network.newNetwork(); + + if (fixedPorts) { + namenode = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(NN_RPC_PORT, NN_RPC_PORT) + .withFixedExposedPort(NN_HTTP_PORT, NN_HTTP_PORT) + .withNetwork(network) + .withNetworkAliases("namenode") + .withCommand("hdfs", "namenode") + .withEnv(COMMON_ENV) + .withEnv("ENSURE_NAMENODE_DIR", "/tmp/hadoop-hadoop/dfs/name") + .waitingFor(Wait.forHttp("/").forPort(NN_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + resourcemanager = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(RM_RPC_PORT, RM_RPC_PORT) + .withFixedExposedPort(RM_HTTP_PORT, RM_HTTP_PORT) + .withNetwork(network) + .withNetworkAliases("resourcemanager") + .withCommand("yarn", "resourcemanager") + .withEnv(COMMON_ENV) + .waitingFor(Wait.forHttp("/ws/v1/cluster/info").forPort(RM_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + datanode = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(DN_XFER_PORT, DN_XFER_PORT) + .withFixedExposedPort(DN_HTTP_PORT, DN_HTTP_PORT) + .withNetwork(network) + .withNetworkAliases("datanode") + .withCommand("hdfs", "datanode") + .withEnv(COMMON_ENV); + } else { + namenode = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("namenode") + .withCommand("hdfs", "namenode") + .withEnv(COMMON_ENV) + .withEnv("ENSURE_NAMENODE_DIR", "/tmp/hadoop-hadoop/dfs/name") + .withExposedPorts(NN_HTTP_PORT, NN_RPC_PORT) + .waitingFor(Wait.forHttp("/").forPort(NN_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + resourcemanager = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("resourcemanager") + .withCommand("yarn", "resourcemanager") + .withEnv(COMMON_ENV) + .withExposedPorts(RM_HTTP_PORT, RM_RPC_PORT) + .waitingFor(Wait.forHttp("/ws/v1/cluster/info").forPort(RM_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + datanode = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("datanode") + .withCommand("hdfs", "datanode") + .withEnv(COMMON_ENV); + } + + if (fixedPorts) { + // Fixed hostname "nodemanager" and published AM_CLIENT_PORT so host JVM can reach the Tez AM. + nodemanager = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(AM_CLIENT_PORT, AM_CLIENT_PORT) + .withCreateContainerCmdModifier(cmd -> cmd.withHostName("nodemanager")) + .withNetwork(network) + .withNetworkAliases("nodemanager") + .withCommand("yarn", "nodemanager") + .withEnv(COMMON_ENV); + } else { + nodemanager = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("nodemanager") + .withCommand("yarn", "nodemanager") + .withEnv(COMMON_ENV); + } + } + + public void start() { + namenode.start(); + datanode.start(); + resourcemanager.start(); + nodemanager.start(); + waitForNodeManagerRegistration(); + verifyJava21InNodeManager(); + } + + private void verifyJava21InNodeManager() { + try { + GenericContainer.ExecResult r = nodemanager.execInContainer( + CONTAINER_JAVA_21_HOME + "/bin/java", "-version"); + if (r.getExitCode() == 0) { + LOG.info("Java 21 is functional in NodeManager ({}): {}", + CONTAINER_JAVA_21_HOME, r.getStderr().trim()); + } else { + LOG.warn("Java 21 check FAILED in NodeManager (exit {}). " + + "Tez AM/task containers will fail at launch time. " + + "stderr: {}", r.getExitCode(), r.getStderr()); + } + } catch (Exception e) { + LOG.warn("Could not verify Java 21 in NodeManager container", e); + } + } + + public void stop() { + nodemanager.stop(); + resourcemanager.stop(); + datanode.stop(); + namenode.stop(); + network.close(); + } + + public String getHdfsUri() { + if (fixedPorts) { + return "hdfs://namenode:" + NN_RPC_PORT; + } + return "hdfs://" + namenode.getHost() + ":" + namenode.getMappedPort(NN_RPC_PORT); + } + + public String getResourceManagerAddress() { + if (fixedPorts) { + return "resourcemanager:" + RM_RPC_PORT; + } + return resourcemanager.getHost() + ":" + resourcemanager.getMappedPort(RM_RPC_PORT); + } + + public String getResourceManagerWebAppAddress() { + if (fixedPorts) { + return "resourcemanager:" + RM_HTTP_PORT; + } + return resourcemanager.getHost() + ":" + resourcemanager.getMappedPort(RM_HTTP_PORT); + } + + public String uploadJarToHdfs(Path localJarPath) throws IOException, InterruptedException { + String fileName = localJarPath.getFileName().toString(); + String containerTmp = "/tmp/" + fileName; + String hdfsDir = "/tmp/hive-29483-jars"; + String hdfsPath = hdfsDir + "/" + fileName; + + namenode.copyFileToContainer(MountableFile.forHostPath(localJarPath, 0644), containerTmp); + + GenericContainer.ExecResult mkdir = namenode.execInContainer("hdfs", "dfs", "-mkdir", "-p", hdfsDir); + requireSuccess(mkdir, "hdfs dfs -mkdir -p " + hdfsDir); + + GenericContainer.ExecResult put = namenode.execInContainer("hdfs", "dfs", "-put", "-f", containerTmp, hdfsPath); + requireSuccess(put, "hdfs dfs -put -f " + containerTmp + " " + hdfsPath); + + return hdfsPath; + } + + public String uploadTezLibsToHdfs() throws IOException, InterruptedException { + String hdfsDir = "/tmp/hive-29483/tez-libs"; + GenericContainer.ExecResult mkdir = namenode.execInContainer("hdfs", "dfs", "-mkdir", "-p", hdfsDir); + requireSuccess(mkdir, "hdfs dfs -mkdir -p " + hdfsDir); + + Set<Path> tezJars = findTezJarsFromClasspath(); + List<String> hdfsUris = new ArrayList<>(); + for (Path jarPath : tezJars) { + String jarName = jarPath.getFileName().toString(); + String containerTmp = "/tmp/" + jarName; + String hdfsPath = hdfsDir + "/" + jarName; + + namenode.copyFileToContainer(MountableFile.forHostPath(jarPath, 0644), containerTmp); + GenericContainer.ExecResult put = namenode.execInContainer( + "hdfs", "dfs", "-put", "-f", containerTmp, hdfsPath); + requireSuccess(put, "hdfs dfs -put -f " + containerTmp + " " + hdfsPath); + + hdfsUris.add("hdfs://namenode:" + NN_RPC_PORT + hdfsPath); + } + + if (hdfsUris.isEmpty()) { + throw new IllegalStateException( + "No framework jars were found on the test classpath to stage. " + + "Ensure tez-api, tez-dag, tez-runtime-library, hadoop-mapreduce-client-core, " + + "etc. are test dependencies."); + } + return String.join(",", hdfsUris); + } + + /** Discovers Tez jars from the classpath via probe-class reflection and string scanning; includes hadoop-shim and MR client jars absent from the cluster classpath. */ + private static Set<Path> findTezJarsFromClasspath() { + Set<Path> jars = new LinkedHashSet<>(); + + String[] probeClassNames = { + "org.apache.tez.dag.api.TezConfiguration", + "org.apache.tez.common.TezConverterUtils", + "org.apache.tez.dag.app.DAGAppMaster", + "org.apache.tez.mapreduce.hadoop.MRHelpers", + "org.apache.tez.runtime.LogicalIOProcessorRuntimeTask", + "org.apache.tez.runtime.library.api.KeyValueReader", + "org.apache.tez.hadoop.shim.HadoopShimsLoader", + "org.apache.hadoop.mapred.JobConf", + "org.apache.hadoop.mapreduce.v2.util.MRApps" + }; + for (String className : probeClassNames) { + try { + Class<?> cls = Class.forName(className); + CodeSource cs = cls.getProtectionDomain().getCodeSource(); + if (cs != null && cs.getLocation() != null) { + String path = cs.getLocation().getPath(); + if (path.endsWith(".jar")) { + Path p = Paths.get(path); + if (isTezFrameworkJar(p.getFileName().toString()) && Files.isRegularFile(p)) { + jars.add(p); + } + } + } + } catch (ClassNotFoundException | SecurityException ignored) { + } + } + + String cp = ManagementFactory.getRuntimeMXBean().getClassPath(); + for (String entry : cp.split(File.pathSeparator)) { + if (!entry.endsWith(".jar")) { + continue; + } + Path p = Paths.get(entry); + if (isTezFrameworkJar(p.getFileName().toString()) && Files.isRegularFile(p)) { + jars.add(p); + } + } + + return jars; + } + + private static boolean isTezFrameworkJar(String name) { + if (name.endsWith("-tests.jar")) { + return false; + } + return name.contains("tez") + || name.startsWith("hadoop-shim") + || name.startsWith("hadoop-mapreduce-client-core") + || name.startsWith("hadoop-mapreduce-client-common"); + } + + GenericContainer<?> namenodeContainer() { + return namenode; + } + + GenericContainer<?> resourceManagerContainer() { + return resourcemanager; + } + + GenericContainer<?> nodeManagerContainer() { + return nodemanager; + } + + private void waitForNodeManagerRegistration() { Review Comment: consider using [Awaitility](https://github.com/awaitility/awaitility) for this purpose, which is already used in the hive project ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestTezYarnLocalization.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class TestTezYarnLocalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestTezYarnLocalization.class); + + private static final String HDFS_BASE = "hdfs://namenode:8020"; + private static final String HDFS_WAREHOUSE = HDFS_BASE + "/tmp/hive-tez-loc/warehouse"; + private static final String HDFS_SCRATCH = HDFS_BASE + "/tmp/hive-tez-loc/scratch"; + private static final String HDFS_ROOT = "/tmp/hive-tez-loc"; + + private static TezYarnClusterContainer cluster; + private static HiveServer2 hs2; + private static int hs2Port; + + @BeforeClass + public static void startAll() throws Exception { + cluster = new TezYarnClusterContainer(true); + cluster.start(); + + GenericContainer<?> nn = cluster.namenodeContainer(); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", "/tmp"); + + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/warehouse"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/scratch"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/user-install"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", HDFS_ROOT); + + String tezLibUris = cluster.uploadTezLibsToHdfs(); + LOG.info("Staged Tez libs to HDFS: {}", tezLibUris); + + Path localScratch = Files.createDirectories( + Path.of("/tmp", "hive-tez-loc-" + System.currentTimeMillis())); + HiveConf conf = buildHiveConf(tezLibUris, localScratch); + + hs2 = new HiveServer2(); + hs2.init(conf); + hs2.start(); + + waitForJdbc(hs2Port); + LOG.info("HiveServer2 is ready on port {}", hs2Port); + } + + @AfterClass + public static void stopAll() { + dumpNodeManagerDiagnostics(); + if (hs2 != null) { + hs2.stop(); + hs2 = null; + } + if (cluster != null) { + cluster.stop(); + cluster = null; + } + } + + @Test + public void testQuerySucceedsWithAppJar() throws Exception { + String url = jdbcUrl(hs2Port); + try (Connection conn = DriverManager.getConnection(url, "hive", "")) { + try (Statement stmt = conn.createStatement()) { + + stmt.execute("CREATE TABLE IF NOT EXISTS tez_loc_test (id INT) STORED AS ORC"); + stmt.execute("CREATE TABLE IF NOT EXISTS tez_source (id INT) STORED AS ORC"); + + stmt.execute("INSERT INTO tez_loc_test SELECT count(*) FROM tez_source"); + + try (ResultSet rs = stmt.executeQuery("SELECT id FROM tez_loc_test")) { + Assert.assertTrue("Result set must contain at least one row", rs.next()); + long count = rs.getLong(1); + Assert.assertEquals( + "INSERT SELECT count(*) FROM empty tez_source should return 0 (hive-exec.jar was localized)", + 0L, count); + LOG.info("Tez query succeeded: inserted count(*) = {}", count); + } + } + } + + verifyTezYarnAppExists(); + } + + /** Prints Tez AM and NodeManager logs to the Surefire *-output.txt file at teardown. */ + private static void dumpNodeManagerDiagnostics() { + if (cluster == null) { + return; + } + System.out.println("########## BEGIN NodeManager diagnostics ##########"); + try { + dumpNmCommand("launch_container.sh (AM launch command + classpath)", + "find /tmp -name 'launch_container.sh' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container syslog (Tez AM log4j output)", + "find /var/log/hadoop/userlogs -name 'syslog*' 2>/dev/null | head -10 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container stdout + stderr + prelaunch.err", + "find /var/log/hadoop/userlogs \\( -name 'stdout' -o -name 'stderr' -o -name 'prelaunch.err' \\) " + + "2>/dev/null | head -20 | xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("NodeManager daemon log (tail 200)", + "find /var/log/hadoop -maxdepth 1 -name '*.log' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; tail -200 {}' 2>/dev/null || true"); + } catch (Exception e) { + System.out.println("Could not dump NodeManager diagnostics: " + e); + } + System.out.println("########## END NodeManager diagnostics ##########"); + System.out.flush(); + } + + private static void dumpNmCommand(String label, String bashCommand) { + try { + GenericContainer.ExecResult r = + cluster.nodeManagerContainer().execInContainer("bash", "-c", bashCommand); + String out = r.getStdout(); + System.out.println("===== NM: " + label + " ====="); + System.out.println(out.isEmpty() ? "(no output found)" : out); + } catch (Exception e) { + System.out.println("===== NM: " + label + " (dump failed: " + e + ") ====="); Review Comment: use logging library ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestTezYarnLocalization.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class TestTezYarnLocalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestTezYarnLocalization.class); + + private static final String HDFS_BASE = "hdfs://namenode:8020"; + private static final String HDFS_WAREHOUSE = HDFS_BASE + "/tmp/hive-tez-loc/warehouse"; + private static final String HDFS_SCRATCH = HDFS_BASE + "/tmp/hive-tez-loc/scratch"; + private static final String HDFS_ROOT = "/tmp/hive-tez-loc"; + + private static TezYarnClusterContainer cluster; + private static HiveServer2 hs2; + private static int hs2Port; + + @BeforeClass + public static void startAll() throws Exception { + cluster = new TezYarnClusterContainer(true); + cluster.start(); + + GenericContainer<?> nn = cluster.namenodeContainer(); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", "/tmp"); + + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/warehouse"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/scratch"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/user-install"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", HDFS_ROOT); + + String tezLibUris = cluster.uploadTezLibsToHdfs(); + LOG.info("Staged Tez libs to HDFS: {}", tezLibUris); + + Path localScratch = Files.createDirectories( + Path.of("/tmp", "hive-tez-loc-" + System.currentTimeMillis())); + HiveConf conf = buildHiveConf(tezLibUris, localScratch); + + hs2 = new HiveServer2(); + hs2.init(conf); + hs2.start(); + + waitForJdbc(hs2Port); + LOG.info("HiveServer2 is ready on port {}", hs2Port); + } + + @AfterClass + public static void stopAll() { + dumpNodeManagerDiagnostics(); + if (hs2 != null) { + hs2.stop(); + hs2 = null; + } + if (cluster != null) { + cluster.stop(); + cluster = null; + } + } + + @Test + public void testQuerySucceedsWithAppJar() throws Exception { + String url = jdbcUrl(hs2Port); + try (Connection conn = DriverManager.getConnection(url, "hive", "")) { + try (Statement stmt = conn.createStatement()) { + + stmt.execute("CREATE TABLE IF NOT EXISTS tez_loc_test (id INT) STORED AS ORC"); + stmt.execute("CREATE TABLE IF NOT EXISTS tez_source (id INT) STORED AS ORC"); + + stmt.execute("INSERT INTO tez_loc_test SELECT count(*) FROM tez_source"); + + try (ResultSet rs = stmt.executeQuery("SELECT id FROM tez_loc_test")) { + Assert.assertTrue("Result set must contain at least one row", rs.next()); + long count = rs.getLong(1); + Assert.assertEquals( + "INSERT SELECT count(*) FROM empty tez_source should return 0 (hive-exec.jar was localized)", + 0L, count); + LOG.info("Tez query succeeded: inserted count(*) = {}", count); + } + } + } + + verifyTezYarnAppExists(); + } + + /** Prints Tez AM and NodeManager logs to the Surefire *-output.txt file at teardown. */ + private static void dumpNodeManagerDiagnostics() { + if (cluster == null) { + return; + } + System.out.println("########## BEGIN NodeManager diagnostics ##########"); + try { + dumpNmCommand("launch_container.sh (AM launch command + classpath)", + "find /tmp -name 'launch_container.sh' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container syslog (Tez AM log4j output)", + "find /var/log/hadoop/userlogs -name 'syslog*' 2>/dev/null | head -10 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container stdout + stderr + prelaunch.err", + "find /var/log/hadoop/userlogs \\( -name 'stdout' -o -name 'stderr' -o -name 'prelaunch.err' \\) " + + "2>/dev/null | head -20 | xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("NodeManager daemon log (tail 200)", + "find /var/log/hadoop -maxdepth 1 -name '*.log' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; tail -200 {}' 2>/dev/null || true"); + } catch (Exception e) { + System.out.println("Could not dump NodeManager diagnostics: " + e); Review Comment: use logging library ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TezYarnClusterContainer.java: ########## @@ -0,0 +1,377 @@ +/* + * 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.hive.tez.yarn; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.FixedHostPortGenericContainer; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.images.builder.ImageFromDockerfile; +import org.testcontainers.utility.MountableFile; + +import java.io.File; +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.security.CodeSource; +import java.time.Duration; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Stream; + +public class TezYarnClusterContainer { + + private static final Logger LOG = LoggerFactory.getLogger(TezYarnClusterContainer.class); + + /** Path to the Java 21 runtime inside containers for Tez AM/task launch environments. */ + public static final String CONTAINER_JAVA_21_HOME = "/opt/jdk21"; + + private static final String HADOOP_IMAGE = buildHadoopImage(); + private static final Duration STARTUP_TIMEOUT = Duration.ofMinutes(3); + private static final Map<String, String> COMMON_ENV = loadCommonEnv(); + + private static final int NN_RPC_PORT = 8020; + private static final int NN_HTTP_PORT = 9870; + private static final int RM_RPC_PORT = 8032; + private static final int RM_HTTP_PORT = 8088; + private static final int DN_HTTP_PORT = 9864; + private static final int DN_XFER_PORT = 9866; + // Tez AM client RPC port, published by the NM container so the host JVM can reach the AM. + public static final int AM_CLIENT_PORT = 41000; + + private final Network network; + private final GenericContainer<?> namenode; + private final GenericContainer<?> datanode; + private final GenericContainer<?> resourcemanager; + private final GenericContainer<?> nodemanager; + private final boolean fixedPorts; + + public TezYarnClusterContainer() { + this(false); + } + + public TezYarnClusterContainer(boolean fixedPorts) { + this.fixedPorts = fixedPorts; + network = Network.newNetwork(); + + if (fixedPorts) { + namenode = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(NN_RPC_PORT, NN_RPC_PORT) + .withFixedExposedPort(NN_HTTP_PORT, NN_HTTP_PORT) + .withNetwork(network) + .withNetworkAliases("namenode") + .withCommand("hdfs", "namenode") + .withEnv(COMMON_ENV) + .withEnv("ENSURE_NAMENODE_DIR", "/tmp/hadoop-hadoop/dfs/name") + .waitingFor(Wait.forHttp("/").forPort(NN_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + resourcemanager = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(RM_RPC_PORT, RM_RPC_PORT) + .withFixedExposedPort(RM_HTTP_PORT, RM_HTTP_PORT) + .withNetwork(network) + .withNetworkAliases("resourcemanager") + .withCommand("yarn", "resourcemanager") + .withEnv(COMMON_ENV) + .waitingFor(Wait.forHttp("/ws/v1/cluster/info").forPort(RM_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + datanode = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(DN_XFER_PORT, DN_XFER_PORT) + .withFixedExposedPort(DN_HTTP_PORT, DN_HTTP_PORT) + .withNetwork(network) + .withNetworkAliases("datanode") + .withCommand("hdfs", "datanode") + .withEnv(COMMON_ENV); + } else { + namenode = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("namenode") + .withCommand("hdfs", "namenode") + .withEnv(COMMON_ENV) + .withEnv("ENSURE_NAMENODE_DIR", "/tmp/hadoop-hadoop/dfs/name") + .withExposedPorts(NN_HTTP_PORT, NN_RPC_PORT) + .waitingFor(Wait.forHttp("/").forPort(NN_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + resourcemanager = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("resourcemanager") + .withCommand("yarn", "resourcemanager") + .withEnv(COMMON_ENV) + .withExposedPorts(RM_HTTP_PORT, RM_RPC_PORT) + .waitingFor(Wait.forHttp("/ws/v1/cluster/info").forPort(RM_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + datanode = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("datanode") + .withCommand("hdfs", "datanode") + .withEnv(COMMON_ENV); + } + + if (fixedPorts) { + // Fixed hostname "nodemanager" and published AM_CLIENT_PORT so host JVM can reach the Tez AM. + nodemanager = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(AM_CLIENT_PORT, AM_CLIENT_PORT) + .withCreateContainerCmdModifier(cmd -> cmd.withHostName("nodemanager")) + .withNetwork(network) + .withNetworkAliases("nodemanager") + .withCommand("yarn", "nodemanager") + .withEnv(COMMON_ENV); + } else { + nodemanager = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("nodemanager") + .withCommand("yarn", "nodemanager") + .withEnv(COMMON_ENV); + } + } + + public void start() { + namenode.start(); + datanode.start(); + resourcemanager.start(); + nodemanager.start(); + waitForNodeManagerRegistration(); + verifyJava21InNodeManager(); + } + + private void verifyJava21InNodeManager() { + try { + GenericContainer.ExecResult r = nodemanager.execInContainer( + CONTAINER_JAVA_21_HOME + "/bin/java", "-version"); + if (r.getExitCode() == 0) { + LOG.info("Java 21 is functional in NodeManager ({}): {}", + CONTAINER_JAVA_21_HOME, r.getStderr().trim()); + } else { + LOG.warn("Java 21 check FAILED in NodeManager (exit {}). " + + "Tez AM/task containers will fail at launch time. " + + "stderr: {}", r.getExitCode(), r.getStderr()); + } + } catch (Exception e) { + LOG.warn("Could not verify Java 21 in NodeManager container", e); + } + } + + public void stop() { + nodemanager.stop(); + resourcemanager.stop(); + datanode.stop(); + namenode.stop(); + network.close(); + } + + public String getHdfsUri() { + if (fixedPorts) { + return "hdfs://namenode:" + NN_RPC_PORT; + } + return "hdfs://" + namenode.getHost() + ":" + namenode.getMappedPort(NN_RPC_PORT); + } + + public String getResourceManagerAddress() { + if (fixedPorts) { + return "resourcemanager:" + RM_RPC_PORT; + } + return resourcemanager.getHost() + ":" + resourcemanager.getMappedPort(RM_RPC_PORT); + } + + public String getResourceManagerWebAppAddress() { + if (fixedPorts) { + return "resourcemanager:" + RM_HTTP_PORT; + } + return resourcemanager.getHost() + ":" + resourcemanager.getMappedPort(RM_HTTP_PORT); + } + + public String uploadJarToHdfs(Path localJarPath) throws IOException, InterruptedException { + String fileName = localJarPath.getFileName().toString(); + String containerTmp = "/tmp/" + fileName; + String hdfsDir = "/tmp/hive-29483-jars"; + String hdfsPath = hdfsDir + "/" + fileName; + + namenode.copyFileToContainer(MountableFile.forHostPath(localJarPath, 0644), containerTmp); + + GenericContainer.ExecResult mkdir = namenode.execInContainer("hdfs", "dfs", "-mkdir", "-p", hdfsDir); + requireSuccess(mkdir, "hdfs dfs -mkdir -p " + hdfsDir); + + GenericContainer.ExecResult put = namenode.execInContainer("hdfs", "dfs", "-put", "-f", containerTmp, hdfsPath); + requireSuccess(put, "hdfs dfs -put -f " + containerTmp + " " + hdfsPath); + + return hdfsPath; + } + + public String uploadTezLibsToHdfs() throws IOException, InterruptedException { + String hdfsDir = "/tmp/hive-29483/tez-libs"; + GenericContainer.ExecResult mkdir = namenode.execInContainer("hdfs", "dfs", "-mkdir", "-p", hdfsDir); + requireSuccess(mkdir, "hdfs dfs -mkdir -p " + hdfsDir); + + Set<Path> tezJars = findTezJarsFromClasspath(); + List<String> hdfsUris = new ArrayList<>(); + for (Path jarPath : tezJars) { + String jarName = jarPath.getFileName().toString(); + String containerTmp = "/tmp/" + jarName; + String hdfsPath = hdfsDir + "/" + jarName; + + namenode.copyFileToContainer(MountableFile.forHostPath(jarPath, 0644), containerTmp); + GenericContainer.ExecResult put = namenode.execInContainer( + "hdfs", "dfs", "-put", "-f", containerTmp, hdfsPath); + requireSuccess(put, "hdfs dfs -put -f " + containerTmp + " " + hdfsPath); + + hdfsUris.add("hdfs://namenode:" + NN_RPC_PORT + hdfsPath); + } + + if (hdfsUris.isEmpty()) { + throw new IllegalStateException( + "No framework jars were found on the test classpath to stage. " + + "Ensure tez-api, tez-dag, tez-runtime-library, hadoop-mapreduce-client-core, " + + "etc. are test dependencies."); + } + return String.join(",", hdfsUris); + } + + /** Discovers Tez jars from the classpath via probe-class reflection and string scanning; includes hadoop-shim and MR client jars absent from the cluster classpath. */ + private static Set<Path> findTezJarsFromClasspath() { + Set<Path> jars = new LinkedHashSet<>(); + + String[] probeClassNames = { + "org.apache.tez.dag.api.TezConfiguration", + "org.apache.tez.common.TezConverterUtils", + "org.apache.tez.dag.app.DAGAppMaster", + "org.apache.tez.mapreduce.hadoop.MRHelpers", + "org.apache.tez.runtime.LogicalIOProcessorRuntimeTask", + "org.apache.tez.runtime.library.api.KeyValueReader", + "org.apache.tez.hadoop.shim.HadoopShimsLoader", + "org.apache.hadoop.mapred.JobConf", + "org.apache.hadoop.mapreduce.v2.util.MRApps" + }; + for (String className : probeClassNames) { + try { + Class<?> cls = Class.forName(className); + CodeSource cs = cls.getProtectionDomain().getCodeSource(); + if (cs != null && cs.getLocation() != null) { + String path = cs.getLocation().getPath(); + if (path.endsWith(".jar")) { + Path p = Paths.get(path); + if (isTezFrameworkJar(p.getFileName().toString()) && Files.isRegularFile(p)) { + jars.add(p); + } + } + } + } catch (ClassNotFoundException | SecurityException ignored) { + } + } + + String cp = ManagementFactory.getRuntimeMXBean().getClassPath(); + for (String entry : cp.split(File.pathSeparator)) { + if (!entry.endsWith(".jar")) { + continue; + } + Path p = Paths.get(entry); + if (isTezFrameworkJar(p.getFileName().toString()) && Files.isRegularFile(p)) { + jars.add(p); + } + } + + return jars; + } + + private static boolean isTezFrameworkJar(String name) { + if (name.endsWith("-tests.jar")) { + return false; + } + return name.contains("tez") + || name.startsWith("hadoop-shim") + || name.startsWith("hadoop-mapreduce-client-core") + || name.startsWith("hadoop-mapreduce-client-common"); + } Review Comment: this whole logic is not needed, I believe, if Tez is used from a tez.tar.gz ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestTezYarnLocalization.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class TestTezYarnLocalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestTezYarnLocalization.class); + + private static final String HDFS_BASE = "hdfs://namenode:8020"; + private static final String HDFS_WAREHOUSE = HDFS_BASE + "/tmp/hive-tez-loc/warehouse"; + private static final String HDFS_SCRATCH = HDFS_BASE + "/tmp/hive-tez-loc/scratch"; + private static final String HDFS_ROOT = "/tmp/hive-tez-loc"; + + private static TezYarnClusterContainer cluster; + private static HiveServer2 hs2; + private static int hs2Port; + + @BeforeClass + public static void startAll() throws Exception { + cluster = new TezYarnClusterContainer(true); + cluster.start(); + + GenericContainer<?> nn = cluster.namenodeContainer(); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", "/tmp"); + + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/warehouse"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/scratch"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/user-install"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", HDFS_ROOT); + + String tezLibUris = cluster.uploadTezLibsToHdfs(); + LOG.info("Staged Tez libs to HDFS: {}", tezLibUris); + + Path localScratch = Files.createDirectories( + Path.of("/tmp", "hive-tez-loc-" + System.currentTimeMillis())); + HiveConf conf = buildHiveConf(tezLibUris, localScratch); + + hs2 = new HiveServer2(); + hs2.init(conf); + hs2.start(); + + waitForJdbc(hs2Port); + LOG.info("HiveServer2 is ready on port {}", hs2Port); + } + + @AfterClass + public static void stopAll() { + dumpNodeManagerDiagnostics(); + if (hs2 != null) { + hs2.stop(); + hs2 = null; + } + if (cluster != null) { + cluster.stop(); + cluster = null; + } + } + + @Test + public void testQuerySucceedsWithAppJar() throws Exception { + String url = jdbcUrl(hs2Port); + try (Connection conn = DriverManager.getConnection(url, "hive", "")) { + try (Statement stmt = conn.createStatement()) { + + stmt.execute("CREATE TABLE IF NOT EXISTS tez_loc_test (id INT) STORED AS ORC"); + stmt.execute("CREATE TABLE IF NOT EXISTS tez_source (id INT) STORED AS ORC"); + + stmt.execute("INSERT INTO tez_loc_test SELECT count(*) FROM tez_source"); + + try (ResultSet rs = stmt.executeQuery("SELECT id FROM tez_loc_test")) { + Assert.assertTrue("Result set must contain at least one row", rs.next()); + long count = rs.getLong(1); + Assert.assertEquals( + "INSERT SELECT count(*) FROM empty tez_source should return 0 (hive-exec.jar was localized)", + 0L, count); + LOG.info("Tez query succeeded: inserted count(*) = {}", count); + } + } + } + + verifyTezYarnAppExists(); + } + + /** Prints Tez AM and NodeManager logs to the Surefire *-output.txt file at teardown. */ + private static void dumpNodeManagerDiagnostics() { + if (cluster == null) { + return; + } + System.out.println("########## BEGIN NodeManager diagnostics ##########"); + try { + dumpNmCommand("launch_container.sh (AM launch command + classpath)", + "find /tmp -name 'launch_container.sh' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container syslog (Tez AM log4j output)", + "find /var/log/hadoop/userlogs -name 'syslog*' 2>/dev/null | head -10 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container stdout + stderr + prelaunch.err", + "find /var/log/hadoop/userlogs \\( -name 'stdout' -o -name 'stderr' -o -name 'prelaunch.err' \\) " + + "2>/dev/null | head -20 | xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("NodeManager daemon log (tail 200)", + "find /var/log/hadoop -maxdepth 1 -name '*.log' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; tail -200 {}' 2>/dev/null || true"); + } catch (Exception e) { + System.out.println("Could not dump NodeManager diagnostics: " + e); + } + System.out.println("########## END NodeManager diagnostics ##########"); + System.out.flush(); Review Comment: use logging library ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestTezYarnLocalization.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class TestTezYarnLocalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestTezYarnLocalization.class); + + private static final String HDFS_BASE = "hdfs://namenode:8020"; + private static final String HDFS_WAREHOUSE = HDFS_BASE + "/tmp/hive-tez-loc/warehouse"; + private static final String HDFS_SCRATCH = HDFS_BASE + "/tmp/hive-tez-loc/scratch"; + private static final String HDFS_ROOT = "/tmp/hive-tez-loc"; + + private static TezYarnClusterContainer cluster; + private static HiveServer2 hs2; + private static int hs2Port; + + @BeforeClass + public static void startAll() throws Exception { + cluster = new TezYarnClusterContainer(true); + cluster.start(); + + GenericContainer<?> nn = cluster.namenodeContainer(); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", "/tmp"); + + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/warehouse"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/scratch"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/user-install"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", HDFS_ROOT); + + String tezLibUris = cluster.uploadTezLibsToHdfs(); + LOG.info("Staged Tez libs to HDFS: {}", tezLibUris); + + Path localScratch = Files.createDirectories( + Path.of("/tmp", "hive-tez-loc-" + System.currentTimeMillis())); + HiveConf conf = buildHiveConf(tezLibUris, localScratch); + + hs2 = new HiveServer2(); + hs2.init(conf); + hs2.start(); + + waitForJdbc(hs2Port); + LOG.info("HiveServer2 is ready on port {}", hs2Port); + } + + @AfterClass + public static void stopAll() { + dumpNodeManagerDiagnostics(); + if (hs2 != null) { + hs2.stop(); + hs2 = null; + } + if (cluster != null) { + cluster.stop(); + cluster = null; + } + } + + @Test + public void testQuerySucceedsWithAppJar() throws Exception { + String url = jdbcUrl(hs2Port); + try (Connection conn = DriverManager.getConnection(url, "hive", "")) { + try (Statement stmt = conn.createStatement()) { + + stmt.execute("CREATE TABLE IF NOT EXISTS tez_loc_test (id INT) STORED AS ORC"); + stmt.execute("CREATE TABLE IF NOT EXISTS tez_source (id INT) STORED AS ORC"); + + stmt.execute("INSERT INTO tez_loc_test SELECT count(*) FROM tez_source"); + + try (ResultSet rs = stmt.executeQuery("SELECT id FROM tez_loc_test")) { + Assert.assertTrue("Result set must contain at least one row", rs.next()); + long count = rs.getLong(1); + Assert.assertEquals( + "INSERT SELECT count(*) FROM empty tez_source should return 0 (hive-exec.jar was localized)", + 0L, count); + LOG.info("Tez query succeeded: inserted count(*) = {}", count); + } + } + } + + verifyTezYarnAppExists(); + } + + /** Prints Tez AM and NodeManager logs to the Surefire *-output.txt file at teardown. */ + private static void dumpNodeManagerDiagnostics() { + if (cluster == null) { + return; + } + System.out.println("########## BEGIN NodeManager diagnostics ##########"); + try { + dumpNmCommand("launch_container.sh (AM launch command + classpath)", + "find /tmp -name 'launch_container.sh' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container syslog (Tez AM log4j output)", + "find /var/log/hadoop/userlogs -name 'syslog*' 2>/dev/null | head -10 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container stdout + stderr + prelaunch.err", + "find /var/log/hadoop/userlogs \\( -name 'stdout' -o -name 'stderr' -o -name 'prelaunch.err' \\) " + + "2>/dev/null | head -20 | xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("NodeManager daemon log (tail 200)", + "find /var/log/hadoop -maxdepth 1 -name '*.log' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; tail -200 {}' 2>/dev/null || true"); Review Comment: after a successful run of `TestTezYarnLocalization` I saw empty nodemanager logs, is this expected? ``` ===== NM: NodeManager daemon log (tail 200) ===== (no output found) ########## END NodeManager diagnostics ########## ``` ########## itests/tez-yarn-it/src/test/docker/hadoop-yarn/config: ########## Review Comment: leaving a generic comment here about this PR: I can really use a readme.md containing useful information about the following: 1. command example on how to start this cluster other than the unit test class and make keep alive for testing 2. beeline command about how to connect to hiveserver2 3. if I change something in hive-exec code, what steps are needed to re-deploy and test it with this dockerized setup (if possible at all) ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TezYarnClusterContainer.java: ########## @@ -0,0 +1,377 @@ +/* + * 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.hive.tez.yarn; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.FixedHostPortGenericContainer; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.images.builder.ImageFromDockerfile; +import org.testcontainers.utility.MountableFile; + +import java.io.File; +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.security.CodeSource; +import java.time.Duration; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Stream; + +public class TezYarnClusterContainer { + + private static final Logger LOG = LoggerFactory.getLogger(TezYarnClusterContainer.class); + + /** Path to the Java 21 runtime inside containers for Tez AM/task launch environments. */ + public static final String CONTAINER_JAVA_21_HOME = "/opt/jdk21"; + + private static final String HADOOP_IMAGE = buildHadoopImage(); + private static final Duration STARTUP_TIMEOUT = Duration.ofMinutes(3); + private static final Map<String, String> COMMON_ENV = loadCommonEnv(); + + private static final int NN_RPC_PORT = 8020; + private static final int NN_HTTP_PORT = 9870; + private static final int RM_RPC_PORT = 8032; + private static final int RM_HTTP_PORT = 8088; + private static final int DN_HTTP_PORT = 9864; + private static final int DN_XFER_PORT = 9866; + // Tez AM client RPC port, published by the NM container so the host JVM can reach the AM. + public static final int AM_CLIENT_PORT = 41000; + + private final Network network; + private final GenericContainer<?> namenode; + private final GenericContainer<?> datanode; + private final GenericContainer<?> resourcemanager; + private final GenericContainer<?> nodemanager; + private final boolean fixedPorts; + + public TezYarnClusterContainer() { + this(false); + } + + public TezYarnClusterContainer(boolean fixedPorts) { + this.fixedPorts = fixedPorts; + network = Network.newNetwork(); + + if (fixedPorts) { + namenode = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(NN_RPC_PORT, NN_RPC_PORT) + .withFixedExposedPort(NN_HTTP_PORT, NN_HTTP_PORT) + .withNetwork(network) + .withNetworkAliases("namenode") + .withCommand("hdfs", "namenode") + .withEnv(COMMON_ENV) + .withEnv("ENSURE_NAMENODE_DIR", "/tmp/hadoop-hadoop/dfs/name") + .waitingFor(Wait.forHttp("/").forPort(NN_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + resourcemanager = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(RM_RPC_PORT, RM_RPC_PORT) + .withFixedExposedPort(RM_HTTP_PORT, RM_HTTP_PORT) + .withNetwork(network) + .withNetworkAliases("resourcemanager") + .withCommand("yarn", "resourcemanager") + .withEnv(COMMON_ENV) + .waitingFor(Wait.forHttp("/ws/v1/cluster/info").forPort(RM_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + datanode = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(DN_XFER_PORT, DN_XFER_PORT) + .withFixedExposedPort(DN_HTTP_PORT, DN_HTTP_PORT) + .withNetwork(network) + .withNetworkAliases("datanode") + .withCommand("hdfs", "datanode") + .withEnv(COMMON_ENV); + } else { + namenode = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("namenode") + .withCommand("hdfs", "namenode") + .withEnv(COMMON_ENV) + .withEnv("ENSURE_NAMENODE_DIR", "/tmp/hadoop-hadoop/dfs/name") + .withExposedPorts(NN_HTTP_PORT, NN_RPC_PORT) + .waitingFor(Wait.forHttp("/").forPort(NN_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + resourcemanager = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("resourcemanager") + .withCommand("yarn", "resourcemanager") + .withEnv(COMMON_ENV) + .withExposedPorts(RM_HTTP_PORT, RM_RPC_PORT) + .waitingFor(Wait.forHttp("/ws/v1/cluster/info").forPort(RM_HTTP_PORT).withStartupTimeout(STARTUP_TIMEOUT)); + + datanode = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("datanode") + .withCommand("hdfs", "datanode") + .withEnv(COMMON_ENV); + } + + if (fixedPorts) { + // Fixed hostname "nodemanager" and published AM_CLIENT_PORT so host JVM can reach the Tez AM. + nodemanager = new FixedHostPortGenericContainer<>(HADOOP_IMAGE) + .withFixedExposedPort(AM_CLIENT_PORT, AM_CLIENT_PORT) + .withCreateContainerCmdModifier(cmd -> cmd.withHostName("nodemanager")) + .withNetwork(network) + .withNetworkAliases("nodemanager") + .withCommand("yarn", "nodemanager") + .withEnv(COMMON_ENV); + } else { + nodemanager = new GenericContainer<>(HADOOP_IMAGE) + .withNetwork(network) + .withNetworkAliases("nodemanager") + .withCommand("yarn", "nodemanager") + .withEnv(COMMON_ENV); + } + } + + public void start() { + namenode.start(); + datanode.start(); + resourcemanager.start(); + nodemanager.start(); + waitForNodeManagerRegistration(); + verifyJava21InNodeManager(); + } + + private void verifyJava21InNodeManager() { + try { + GenericContainer.ExecResult r = nodemanager.execInContainer( + CONTAINER_JAVA_21_HOME + "/bin/java", "-version"); + if (r.getExitCode() == 0) { + LOG.info("Java 21 is functional in NodeManager ({}): {}", + CONTAINER_JAVA_21_HOME, r.getStderr().trim()); + } else { + LOG.warn("Java 21 check FAILED in NodeManager (exit {}). " + + "Tez AM/task containers will fail at launch time. " + + "stderr: {}", r.getExitCode(), r.getStderr()); + } + } catch (Exception e) { + LOG.warn("Could not verify Java 21 in NodeManager container", e); + } + } + + public void stop() { + nodemanager.stop(); + resourcemanager.stop(); + datanode.stop(); + namenode.stop(); + network.close(); + } + + public String getHdfsUri() { + if (fixedPorts) { + return "hdfs://namenode:" + NN_RPC_PORT; + } + return "hdfs://" + namenode.getHost() + ":" + namenode.getMappedPort(NN_RPC_PORT); + } + + public String getResourceManagerAddress() { + if (fixedPorts) { + return "resourcemanager:" + RM_RPC_PORT; + } + return resourcemanager.getHost() + ":" + resourcemanager.getMappedPort(RM_RPC_PORT); + } + + public String getResourceManagerWebAppAddress() { + if (fixedPorts) { + return "resourcemanager:" + RM_HTTP_PORT; + } + return resourcemanager.getHost() + ":" + resourcemanager.getMappedPort(RM_HTTP_PORT); + } + + public String uploadJarToHdfs(Path localJarPath) throws IOException, InterruptedException { + String fileName = localJarPath.getFileName().toString(); + String containerTmp = "/tmp/" + fileName; + String hdfsDir = "/tmp/hive-29483-jars"; + String hdfsPath = hdfsDir + "/" + fileName; + + namenode.copyFileToContainer(MountableFile.forHostPath(localJarPath, 0644), containerTmp); + + GenericContainer.ExecResult mkdir = namenode.execInContainer("hdfs", "dfs", "-mkdir", "-p", hdfsDir); + requireSuccess(mkdir, "hdfs dfs -mkdir -p " + hdfsDir); + + GenericContainer.ExecResult put = namenode.execInContainer("hdfs", "dfs", "-put", "-f", containerTmp, hdfsPath); + requireSuccess(put, "hdfs dfs -put -f " + containerTmp + " " + hdfsPath); + + return hdfsPath; + } + + public String uploadTezLibsToHdfs() throws IOException, InterruptedException { Review Comment: This class loads Tez jars from the classpath, which looks too complicated: instead, as Tez is supposed to be usable from the tez.tar.gz artifact it releases, it should be downloaded and pointes by the `tez.lib.uris` config: this makes the testing setup even more like a real cluster ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TestTezYarnLocalization.java: ########## @@ -0,0 +1,262 @@ +/* + * 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.hive.tez.yarn; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hive.service.server.HiveServer2; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.GenericContainer; + +import java.net.ServerSocket; +import java.net.URL; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +public class TestTezYarnLocalization { + + private static final Logger LOG = LoggerFactory.getLogger(TestTezYarnLocalization.class); + + private static final String HDFS_BASE = "hdfs://namenode:8020"; + private static final String HDFS_WAREHOUSE = HDFS_BASE + "/tmp/hive-tez-loc/warehouse"; + private static final String HDFS_SCRATCH = HDFS_BASE + "/tmp/hive-tez-loc/scratch"; + private static final String HDFS_ROOT = "/tmp/hive-tez-loc"; + + private static TezYarnClusterContainer cluster; + private static HiveServer2 hs2; + private static int hs2Port; + + @BeforeClass + public static void startAll() throws Exception { + cluster = new TezYarnClusterContainer(true); + cluster.start(); + + GenericContainer<?> nn = cluster.namenodeContainer(); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", "/tmp"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", "/tmp"); + + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/warehouse"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/scratch"); + nn.execInContainer("hdfs", "dfs", "-mkdir", "-p", HDFS_ROOT + "/user-install"); + nn.execInContainer("hdfs", "dfs", "-chmod", "-R", "777", HDFS_ROOT); + + String tezLibUris = cluster.uploadTezLibsToHdfs(); + LOG.info("Staged Tez libs to HDFS: {}", tezLibUris); + + Path localScratch = Files.createDirectories( + Path.of("/tmp", "hive-tez-loc-" + System.currentTimeMillis())); + HiveConf conf = buildHiveConf(tezLibUris, localScratch); + + hs2 = new HiveServer2(); + hs2.init(conf); + hs2.start(); + + waitForJdbc(hs2Port); + LOG.info("HiveServer2 is ready on port {}", hs2Port); + } + + @AfterClass + public static void stopAll() { + dumpNodeManagerDiagnostics(); + if (hs2 != null) { + hs2.stop(); + hs2 = null; + } + if (cluster != null) { + cluster.stop(); + cluster = null; + } + } + + @Test + public void testQuerySucceedsWithAppJar() throws Exception { + String url = jdbcUrl(hs2Port); + try (Connection conn = DriverManager.getConnection(url, "hive", "")) { + try (Statement stmt = conn.createStatement()) { + + stmt.execute("CREATE TABLE IF NOT EXISTS tez_loc_test (id INT) STORED AS ORC"); + stmt.execute("CREATE TABLE IF NOT EXISTS tez_source (id INT) STORED AS ORC"); + + stmt.execute("INSERT INTO tez_loc_test SELECT count(*) FROM tez_source"); + + try (ResultSet rs = stmt.executeQuery("SELECT id FROM tez_loc_test")) { + Assert.assertTrue("Result set must contain at least one row", rs.next()); + long count = rs.getLong(1); + Assert.assertEquals( + "INSERT SELECT count(*) FROM empty tez_source should return 0 (hive-exec.jar was localized)", + 0L, count); + LOG.info("Tez query succeeded: inserted count(*) = {}", count); + } + } + } + + verifyTezYarnAppExists(); + } + + /** Prints Tez AM and NodeManager logs to the Surefire *-output.txt file at teardown. */ + private static void dumpNodeManagerDiagnostics() { + if (cluster == null) { + return; + } + System.out.println("########## BEGIN NodeManager diagnostics ##########"); + try { + dumpNmCommand("launch_container.sh (AM launch command + classpath)", + "find /tmp -name 'launch_container.sh' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container syslog (Tez AM log4j output)", + "find /var/log/hadoop/userlogs -name 'syslog*' 2>/dev/null | head -10 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("container stdout + stderr + prelaunch.err", + "find /var/log/hadoop/userlogs \\( -name 'stdout' -o -name 'stderr' -o -name 'prelaunch.err' \\) " + + "2>/dev/null | head -20 | xargs -I{} sh -c 'echo \"--- {} ---\"; cat {}' 2>/dev/null || true"); + + dumpNmCommand("NodeManager daemon log (tail 200)", + "find /var/log/hadoop -maxdepth 1 -name '*.log' 2>/dev/null | head -3 " + + "| xargs -I{} sh -c 'echo \"--- {} ---\"; tail -200 {}' 2>/dev/null || true"); + } catch (Exception e) { + System.out.println("Could not dump NodeManager diagnostics: " + e); + } + System.out.println("########## END NodeManager diagnostics ##########"); + System.out.flush(); + } + + private static void dumpNmCommand(String label, String bashCommand) { + try { + GenericContainer.ExecResult r = + cluster.nodeManagerContainer().execInContainer("bash", "-c", bashCommand); + String out = r.getStdout(); + System.out.println("===== NM: " + label + " ====="); + System.out.println(out.isEmpty() ? "(no output found)" : out); + } catch (Exception e) { + System.out.println("===== NM: " + label + " (dump failed: " + e + ") ====="); + } + } + + private static HiveConf buildHiveConf(String tezLibUris, Path localScratch) throws Exception { + HiveConf conf = new HiveConf(); + + URL hiveSite = TestTezYarnLocalization.class.getClassLoader().getResource("hive-site-yarn-it.xml"); + URL yarnSite = TestTezYarnLocalization.class.getClassLoader().getResource("yarn-site.xml"); + if (hiveSite != null) { conf.addResource(hiveSite); } + if (yarnSite != null) { conf.addResource(yarnSite); } + + conf.set("fs.defaultFS", HDFS_BASE); + conf.setBoolean("dfs.client.use.datanode.hostname", true); + conf.set("hive.metastore.warehouse.dir", HDFS_WAREHOUSE); + conf.set(HiveConf.ConfVars.SCRATCH_DIR.varname, HDFS_SCRATCH); + conf.set(HiveConf.ConfVars.LOCAL_SCRATCH_DIR.varname, localScratch.toAbsolutePath().toString()); + conf.setVar(HiveConf.ConfVars.HIVE_USER_INSTALL_DIR, HDFS_ROOT + "/user-install"); + + conf.set("javax.jdo.option.ConnectionURL", + "jdbc:derby:" + localScratch.resolve("metastore_db").toAbsolutePath() + ";create=true"); + + conf.setBoolVar(HiveConf.ConfVars.METASTORE_TRY_DIRECT_SQL, false); + conf.set("hive.stats.autogather", "false"); + conf.set("hive.stats.column.autogather", "false"); + conf.set("yarn.resourcemanager.hostname", "resourcemanager"); + conf.set("yarn.resourcemanager.address", "resourcemanager:8032"); + conf.set("yarn.resourcemanager.webapp.address", "resourcemanager:8088"); + + conf.set("tez.lib.uris", tezLibUris); + conf.setBoolean("tez.use.cluster.hadoop-libs", true); + conf.setBoolVar(HiveConf.ConfVars.HIVE_SERVER2_TEZ_INITIALIZE_DEFAULT_SESSIONS, false); + conf.setIntVar(HiveConf.ConfVars.HIVE_SERVER2_TEZ_SESSIONS_PER_DEFAULT_QUEUE, 0); + + conf.set("tez.am.client.am.port-range", + TezYarnClusterContainer.AM_CLIENT_PORT + "-" + TezYarnClusterContainer.AM_CLIENT_PORT); + + String containerEnv = "JAVA_HOME=" + TezYarnClusterContainer.CONTAINER_JAVA_21_HOME + + ",HADOOP_HOME=/opt/hadoop" + + ",HADOOP_MAPRED_HOME=/opt/hadoop"; + conf.set("tez.am.launch.env", containerEnv); + conf.set("tez.task.launch.env", containerEnv); + + hs2Port = findFreePort(); + conf.setIntVar(HiveConf.ConfVars.HIVE_SERVER2_THRIFT_PORT, hs2Port); + conf.setVar(HiveConf.ConfVars.HIVE_SERVER2_THRIFT_BIND_HOST, "localhost"); + conf.setIntVar(HiveConf.ConfVars.HIVE_SERVER2_WEBUI_PORT, findFreePort()); + conf.setVar(HiveConf.ConfVars.HIVE_SERVER2_TRANSPORT_MODE, "binary"); + conf.setVar(HiveConf.ConfVars.HIVE_SERVER2_AUTHENTICATION, "NOSASL"); + conf.setBoolVar(HiveConf.ConfVars.HIVE_SERVER2_ENABLE_DOAS, false); Review Comment: is it possible to move whatever is constant to `hive-site-yarn-it.xml`? ########## itests/tez-yarn-it/src/test/java/org/apache/hive/tez/yarn/TezYarnClusterContainer.java: ########## @@ -0,0 +1,377 @@ +/* + * 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.hive.tez.yarn; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testcontainers.containers.FixedHostPortGenericContainer; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.Network; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.images.builder.ImageFromDockerfile; +import org.testcontainers.utility.MountableFile; + +import java.io.File; +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.security.CodeSource; +import java.time.Duration; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.stream.Stream; + +public class TezYarnClusterContainer { Review Comment: design consideration: would you consider moving all this stuff to a docker-compose file for users' convenience (see my other comment regarding readme: https://github.com/apache/hive/pull/6640/changes#r3842926313) I think this PR now handles a lot of the Java code when it comes to docker setup, which otherwise would be a simple docker-compose and then `TezYarnClusterContainer` can leverage ComposeContainers like: ``` File composeFile = ... "src/test/docker/hadoop-yarn/docker-compose.yml"); ComposeContainer c = new ComposeContainer(composeFile) ``` given that "docker compose" is not available on precommit pods, and we don't want to make it a hard dependency for devs, `withLocalCompose(false)` will work I'm hoping that a docker-compose will save us from using workarounds/hacks like the "custom_hosts_file" for docker compose and the surrounding templates and stuff, feel free to take a look at whatever we already have in the packaging module: https://github.com/apache/hive/blob/master/packaging/src/docker/docker-compose.yml ########## itests/tez-yarn-it/src/test/docker/hadoop-yarn/Dockerfile: ########## @@ -0,0 +1,40 @@ +# 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. + +# Custom Hadoop image for the Tez-on-YARN localization integration tests. +# +# The official apache/hadoop:3.4.2 image ships Java 8 for its HDFS/YARN daemons. +# Hive is compiled for Java 21, so Tez AM and task containers -- which load +# hive-exec classes -- must run on Java 21. This image adds a Java 21 JDK at +# /opt/jdk21 using a multi-stage build from eclipse-temurin:21-jdk; the daemons +# continue to use the default Java 8. +FROM --platform=linux/amd64 eclipse-temurin:21-jdk AS jdk21 +FROM apache/hadoop:3.4.2 + +USER root +COPY --from=jdk21 /opt/java/openjdk /opt/jdk21 +# Ensure the JDK binaries are executable and verify the copy is functional. +# Fails at image-build time if the JDK is incompatible with the base image's +# glibc, surfacing the problem early rather than silently at YARN container +# launch time (where YARN shows exit code 1 and empty stderr). +RUN chmod 755 /opt/jdk21/bin/java && /opt/jdk21/bin/java -version + +# commons-collections 3.2.2 is needed by the Tez AM's HDFS client path in +# Tez 0.10.x; it is not bundled in apache/hadoop:3.4.2. +RUN wget -q https://repo1.maven.org/maven2/commons-collections/commons-collections/3.2.2/commons-collections-3.2.2.jar \ + -P /opt/hadoop/share/hadoop/common/lib/ Review Comment: this is not going to be needed if tez jars are brought from a tez.tar.gz -- 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. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
