You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by se...@apache.org on 2017/09/29 20:56:24 UTC
[1/6] hive git commit: HIVE-17601 : improve error handling in
LlapServiceDriver (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
Repository: hive
Updated Branches:
refs/heads/master b3dbe3136 -> 23478cfeb
HIVE-17601 : improve error handling in LlapServiceDriver (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/23478cfe
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/23478cfe
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/23478cfe
Branch: refs/heads/master
Commit: 23478cfeb9e2cdd676a9612477b85318c59018e1
Parents: 1e4097e
Author: sergey <se...@apache.org>
Authored: Fri Sep 29 13:44:35 2017 -0700
Committer: sergey <se...@apache.org>
Committed: Fri Sep 29 13:56:18 2017 -0700
----------------------------------------------------------------------
.../hadoop/hive/llap/cli/LlapSliderUtils.java | 17 +-
.../llap/cli/LlapStatusOptionsProcessor.java | 33 ++-
.../hive/llap/cli/LlapStatusServiceDriver.java | 206 ++++++++++---------
3 files changed, 135 insertions(+), 121 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/23478cfe/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapSliderUtils.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapSliderUtils.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapSliderUtils.java
index 2d0121c..a0af554 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapSliderUtils.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapSliderUtils.java
@@ -18,12 +18,13 @@
package org.apache.hadoop.hive.llap.cli;
+import com.google.common.io.Files;
import java.io.File;
import java.io.IOException;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.llap.cli.status.LlapStatusHelpers.AppStatusBuilder;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.util.Clock;
@@ -39,8 +40,6 @@ import org.apache.slider.core.exceptions.UnknownApplicationInstanceException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.io.Files;
-
public class LlapSliderUtils {
private static final String SLIDER_GZ = "slider-agent.tar.gz";
private static final Logger LOG = LoggerFactory.getLogger(LlapSliderUtils.class);
@@ -96,7 +95,7 @@ public class LlapSliderUtils {
}
public static ApplicationDiagnostics getApplicationDiagnosticsFromYarnDiagnostics(
- ApplicationReport appReport, Logger LOG) {
+ ApplicationReport appReport, AppStatusBuilder appStatusBuilder, Logger LOG) {
if (appReport == null) {
return null;
}
@@ -105,13 +104,11 @@ public class LlapSliderUtils {
return null;
}
try {
- ApplicationDiagnostics appDiagnostics =
- ApplicationDiagnostics.fromJson(diagnostics);
- return appDiagnostics;
+ return ApplicationDiagnostics.fromJson(diagnostics);
} catch (IOException e) {
- LOG.warn(
- "Failed to parse application diagnostics from Yarn Diagnostics - {}",
- diagnostics);
+ LOG.warn("Failed to parse application diagnostics from Yarn Diagnostics - {}", diagnostics);
+ // Set the raw YARN diagnostics here - the caller won't know if they even exist.
+ appStatusBuilder.setDiagnostics(diagnostics);
return null;
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/23478cfe/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusOptionsProcessor.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusOptionsProcessor.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusOptionsProcessor.java
index bd91495..4c30b80 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusOptionsProcessor.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusOptionsProcessor.java
@@ -39,6 +39,7 @@ public class LlapStatusOptionsProcessor {
private static final long DEFAULT_WATCH_MODE_TIMEOUT_MS = 5 * 60 * 1000l; // 5 minutes timeout for watch mode
private static final float DEFAULT_RUNNING_NODES_THRESHOLD = 1.0f;
+ // TODO: why doesn't this use one of the existing options implementations?!
enum OptionConstants {
NAME("name", 'n', "LLAP cluster name", true),
@@ -48,6 +49,10 @@ public class LlapStatusOptionsProcessor {
OUTPUT_FILE("outputFile", 'o', "File to which output should be written (Default stdout)", true),
WATCH_MODE("watch", 'w', "Watch mode waits until all LLAP daemons are running or subset of the nodes are " +
"running (threshold can be specified via -r option) (Default wait until all nodes are running)", false),
+ // This is a negative because we want the positive to be the default when nothing is specified.
+ NOT_LAUNCHED("notlaunched", 'l', "In watch mode, do not assume that the application was "
+ + "already launched if there's doubt (e.g. if the last application instance has failed).",
+ false),
RUNNING_NODES_THRESHOLD("runningNodesThreshold", 'r', "When watch mode is enabled (-w), wait until the " +
"specified threshold of nodes are running (Default 1.0 which means 100% nodes are running)", true),
STATUS_REFRESH_INTERVAL("refreshInterval", 'i', "Amount of time in seconds to wait until subsequent status checks" +
@@ -108,16 +113,17 @@ public class LlapStatusOptionsProcessor {
private final boolean watchMode;
private final long watchTimeout;
private final float runningNodesThreshold;
+ private final boolean isLaunched;
public LlapStatusOptions(final String name) {
this(name, new Properties(), FIND_YARN_APP_TIMEOUT_MS, null, DEFAULT_STATUS_REFRESH_INTERVAL_MS, false,
- DEFAULT_WATCH_MODE_TIMEOUT_MS, DEFAULT_RUNNING_NODES_THRESHOLD);
+ DEFAULT_WATCH_MODE_TIMEOUT_MS, DEFAULT_RUNNING_NODES_THRESHOLD, true);
}
public LlapStatusOptions(String name, Properties hiveProperties, long findAppTimeoutMs,
String outputFile, long refreshIntervalMs,
final boolean watchMode, final long watchTimeoutMs,
- final float runningNodesThreshold) {
+ final float runningNodesThreshold, final boolean isLaunched) {
this.name = name;
this.conf = hiveProperties;
this.findAppTimeoutMs = findAppTimeoutMs;
@@ -126,6 +132,7 @@ public class LlapStatusOptionsProcessor {
this.watchMode = watchMode;
this.watchTimeout = watchTimeoutMs;
this.runningNodesThreshold = runningNodesThreshold;
+ this.isLaunched = isLaunched;
}
public String getName() {
@@ -152,6 +159,10 @@ public class LlapStatusOptionsProcessor {
return watchMode;
}
+ public boolean isLaunched() {
+ return isLaunched;
+ }
+
public long getWatchTimeoutMs() {
return watchTimeout;
}
@@ -217,26 +228,30 @@ public class LlapStatusOptionsProcessor {
refreshIntervalMs = TimeUnit.MILLISECONDS.convert(refreshIntervalSec, TimeUnit.SECONDS);
}
- boolean watchMode = commandLine.hasOption(OptionConstants.WATCH_MODE.getLongOpt()) ? true : false;
+ boolean watchMode = commandLine.hasOption(OptionConstants.WATCH_MODE.getLongOpt());
long watchTimeoutMs = DEFAULT_WATCH_MODE_TIMEOUT_MS;
if (commandLine.hasOption(OptionConstants.WATCH_MODE_TIMEOUT.getLongOpt())) {
- long watchTimeoutSec = Long.parseLong(commandLine.getOptionValue(OptionConstants.WATCH_MODE_TIMEOUT.getLongOpt()));
+ long watchTimeoutSec = Long.parseLong(commandLine.getOptionValue(
+ OptionConstants.WATCH_MODE_TIMEOUT.getLongOpt()));
if (watchTimeoutSec <= 0) {
throw new IllegalArgumentException("Watch timeout should be >0");
}
watchTimeoutMs = TimeUnit.MILLISECONDS.convert(watchTimeoutSec, TimeUnit.SECONDS);
}
+
+ boolean isLaunched = !commandLine.hasOption(OptionConstants.NOT_LAUNCHED.getLongOpt());
float runningNodesThreshold = DEFAULT_RUNNING_NODES_THRESHOLD;
if (commandLine.hasOption(OptionConstants.RUNNING_NODES_THRESHOLD.getLongOpt())) {
- runningNodesThreshold = Float.parseFloat(commandLine.getOptionValue(OptionConstants.RUNNING_NODES_THRESHOLD
- .getLongOpt()));
+ runningNodesThreshold = Float.parseFloat(commandLine.getOptionValue(
+ OptionConstants.RUNNING_NODES_THRESHOLD.getLongOpt()));
if (runningNodesThreshold < 0.0f || runningNodesThreshold > 1.0f) {
- throw new IllegalArgumentException("Running nodes threshold value should be between 0.0 and 1.0 (inclusive)");
+ throw new IllegalArgumentException(
+ "Running nodes threshold value should be between 0.0 and 1.0 (inclusive)");
}
}
- return new LlapStatusOptions(name, hiveConf, findAppTimeoutMs, outputFile, refreshIntervalMs, watchMode,
- watchTimeoutMs, runningNodesThreshold);
+ return new LlapStatusOptions(name, hiveConf, findAppTimeoutMs, outputFile, refreshIntervalMs,
+ watchMode, watchTimeoutMs, runningNodesThreshold, isLaunched);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/23478cfe/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusServiceDriver.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusServiceDriver.java b/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusServiceDriver.java
index 4fad441..296a851 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusServiceDriver.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/cli/LlapStatusServiceDriver.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.hive.llap.cli;
+import com.google.common.annotations.VisibleForTesting;
import java.io.BufferedOutputStream;
import java.io.FileOutputStream;
import java.io.IOException;
@@ -33,8 +34,6 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
-
-import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
@@ -44,6 +43,7 @@ import org.apache.hadoop.hive.llap.cli.LlapStatusOptionsProcessor.LlapStatusOpti
import org.apache.hadoop.hive.llap.cli.status.LlapStatusHelpers;
import org.apache.hadoop.hive.llap.cli.status.LlapStatusHelpers.AppStatusBuilder;
import org.apache.hadoop.hive.llap.cli.status.LlapStatusHelpers.LlapInstance;
+import org.apache.hadoop.hive.llap.cli.status.LlapStatusHelpers.State;
import org.apache.hadoop.hive.llap.configuration.LlapDaemonConfiguration;
import org.apache.hadoop.hive.llap.registry.LlapServiceInstance;
import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService;
@@ -71,6 +71,10 @@ import org.slf4j.LoggerFactory;
public class LlapStatusServiceDriver {
+ private static final EnumSet<State> NO_SLIDER_INFO_STATES = EnumSet.of(
+ State.APP_NOT_FOUND, State.COMPLETE, State.LAUNCHING);
+ private static final EnumSet<State> LAUNCHING_STATES = EnumSet.of(
+ State.LAUNCHING, State.RUNNING_PARTIAL, State.RUNNING_ALL);
private static final Logger LOG = LoggerFactory.getLogger(LlapStatusServiceDriver.class);
private static final Logger CONSOLE_LOGGER = LoggerFactory.getLogger("LlapStatusServiceDriverConsole");
@@ -252,8 +256,7 @@ public class LlapStatusServiceDriver {
if (ret != ExitCode.SUCCESS) {
return ret.getInt();
- } else if (EnumSet.of(LlapStatusHelpers.State.APP_NOT_FOUND, LlapStatusHelpers.State.COMPLETE, LlapStatusHelpers.State.LAUNCHING)
- .contains(appStatusBuilder.getState())) {
+ } else if (NO_SLIDER_INFO_STATES.contains(appStatusBuilder.getState())) {
return ExitCode.SUCCESS.getInt();
} else {
// Get information from slider.
@@ -375,7 +378,7 @@ public class LlapStatusServiceDriver {
AppStatusBuilder appStatusBuilder) throws
LlapStatusCliException {
if (appReport == null) {
- appStatusBuilder.setState(LlapStatusHelpers.State.APP_NOT_FOUND);
+ appStatusBuilder.setState(State.APP_NOT_FOUND);
LOG.info("No Application Found");
return ExitCode.SUCCESS;
}
@@ -388,11 +391,11 @@ public class LlapStatusServiceDriver {
case NEW:
case NEW_SAVING:
case SUBMITTED:
- appStatusBuilder.setState(LlapStatusHelpers.State.LAUNCHING);
+ appStatusBuilder.setState(State.LAUNCHING);
return ExitCode.SUCCESS;
case ACCEPTED:
appStatusBuilder.maybeCreateAndGetAmInfo().setAppId(appReport.getApplicationId().toString());
- appStatusBuilder.setState(LlapStatusHelpers.State.LAUNCHING);
+ appStatusBuilder.setState(State.LAUNCHING);
return ExitCode.SUCCESS;
case RUNNING:
appStatusBuilder.maybeCreateAndGetAmInfo().setAppId(appReport.getApplicationId().toString());
@@ -403,8 +406,10 @@ public class LlapStatusServiceDriver {
case KILLED:
appStatusBuilder.maybeCreateAndGetAmInfo().setAppId(appReport.getApplicationId().toString());
appStatusBuilder.setAppFinishTime(appReport.getFinishTime());
- appStatusBuilder.setState(LlapStatusHelpers.State.COMPLETE);
- ApplicationDiagnostics appDiagnostics = LlapSliderUtils.getApplicationDiagnosticsFromYarnDiagnostics(appReport, LOG);
+ appStatusBuilder.setState(State.COMPLETE);
+ ApplicationDiagnostics appDiagnostics =
+ LlapSliderUtils.getApplicationDiagnosticsFromYarnDiagnostics(
+ appReport, appStatusBuilder, LOG);
if (appDiagnostics == null) {
LOG.warn("AppDiagnostics not available for YARN application report");
} else {
@@ -575,7 +580,7 @@ public class LlapStatusServiceDriver {
LOG.debug("No information found in the LLAP registry");
}
appStatusBuilder.setLiveInstances(0);
- appStatusBuilder.setState(LlapStatusHelpers.State.LAUNCHING);
+ appStatusBuilder.setState(State.LAUNCHING);
appStatusBuilder.clearRunningLlapInstances();
return ExitCode.SUCCESS;
} else {
@@ -608,15 +613,15 @@ public class LlapStatusServiceDriver {
appStatusBuilder.setLiveInstances(validatedInstances.size());
appStatusBuilder.setLaunchingInstances(llapExtraInstances.size());
if (validatedInstances.size() >= appStatusBuilder.getDesiredInstances()) {
- appStatusBuilder.setState(LlapStatusHelpers.State.RUNNING_ALL);
+ appStatusBuilder.setState(State.RUNNING_ALL);
if (validatedInstances.size() > appStatusBuilder.getDesiredInstances()) {
LOG.warn("Found more entries in LLAP registry, as compared to desired entries");
}
} else {
if (validatedInstances.size() > 0) {
- appStatusBuilder.setState(LlapStatusHelpers.State.RUNNING_PARTIAL);
+ appStatusBuilder.setState(State.RUNNING_PARTIAL);
} else {
- appStatusBuilder.setState(LlapStatusHelpers.State.LAUNCHING);
+ appStatusBuilder.setState(State.LAUNCHING);
}
}
@@ -886,116 +891,113 @@ public class LlapStatusServiceDriver {
TimeUnit.SECONDS.convert(watchTimeout, TimeUnit.MILLISECONDS),
numAttempts, watchMode, new DecimalFormat("#.###").format(runningNodesThreshold));
while (numAttempts > 0) {
- try {
- if (!firstAttempt) {
- if (watchMode) {
- try {
- Thread.sleep(refreshInterval);
- } catch (InterruptedException e) {
- // ignore
- }
- } else {
- // reported once, so break
- break;
+ if (!firstAttempt) {
+ if (watchMode) {
+ try {
+ Thread.sleep(refreshInterval);
+ } catch (InterruptedException e) {
+ // ignore
}
} else {
- firstAttempt = false;
- }
- ret = statusServiceDriver.run(options, watchMode ? watchTimeout : 0);
- currentState = statusServiceDriver.appStatusBuilder.getState();
- try {
- lastSummaryLogTime = LlapStatusServiceDriver
- .maybeLogSummary(clock, lastSummaryLogTime, statusServiceDriver,
- watchMode, watchTimeout, launchingState);
- } catch (Exception e) {
- LOG.warn("Failed to log summary", e);
+ // reported once, so break
+ break;
}
+ } else {
+ firstAttempt = false;
+ }
+ ret = statusServiceDriver.run(options, watchMode ? watchTimeout : 0);
+ currentState = statusServiceDriver.appStatusBuilder.getState();
+ try {
+ lastSummaryLogTime = LlapStatusServiceDriver
+ .maybeLogSummary(clock, lastSummaryLogTime, statusServiceDriver,
+ watchMode, watchTimeout, launchingState);
+ } catch (Exception e) {
+ LOG.warn("Failed to log summary", e);
+ }
- if (ret == ExitCode.SUCCESS.getInt()) {
- if (watchMode) {
+ if (ret == ExitCode.SUCCESS.getInt()) {
+ if (watchMode) {
- // slider has started llap application, now if for some reason state changes to COMPLETE then fail fast
- if (launchingState == null &&
- (EnumSet.of(LlapStatusHelpers.State.LAUNCHING,
- LlapStatusHelpers.State.RUNNING_PARTIAL,
- LlapStatusHelpers.State.RUNNING_ALL)
- .contains(currentState))) {
- launchingState = currentState;
- }
+ // slider has started llap application, now if for some reason state changes to COMPLETE then fail fast
+ if (launchingState == null && LAUNCHING_STATES.contains(currentState)) {
+ launchingState = currentState;
+ }
- if (launchingState != null && currentState.equals(
- LlapStatusHelpers.State.COMPLETE)) {
- LOG.warn("Application stopped while launching. COMPLETE state reached while waiting for RUNNING state."
- + " Failing " + "fast..");
+ if (currentState.equals(State.COMPLETE)) {
+ if (launchingState != null || options.isLaunched()) {
+ LOG.warn("COMPLETE state reached while waiting for RUNNING state. Failing.");
+ System.err.println("Final diagnostics: " +
+ statusServiceDriver.appStatusBuilder.getDiagnostics());
break;
+ } else {
+ LOG.info("Found a stopped application; assuming it was a previous attempt "
+ + "and waiting for the next one. Omit the -l flag to avoid this.");
}
+ }
- if (!(currentState.equals(LlapStatusHelpers.State.RUNNING_PARTIAL) || currentState.equals(
- LlapStatusHelpers.State.RUNNING_ALL))) {
+ if (!(currentState.equals(State.RUNNING_PARTIAL) || currentState.equals(
+ State.RUNNING_ALL))) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Current state: {}. Desired state: {}. {}/{} instances.",
+ currentState,
+ runningNodesThreshold == 1.0f ?
+ State.RUNNING_ALL :
+ State.RUNNING_PARTIAL,
+ statusServiceDriver.appStatusBuilder.getLiveInstances(),
+ statusServiceDriver.appStatusBuilder
+ .getDesiredInstances());
+ }
+ numAttempts--;
+ continue;
+ }
+
+ // we have reached RUNNING state, now check if running nodes threshold is met
+ final int liveInstances = statusServiceDriver.appStatusBuilder.getLiveInstances();
+ final int desiredInstances = statusServiceDriver.appStatusBuilder.getDesiredInstances();
+ if (desiredInstances > 0) {
+ final float ratio = (float) liveInstances / (float) desiredInstances;
+ if (ratio < runningNodesThreshold) {
if (LOG.isDebugEnabled()) {
LOG.debug(
- "Current state: {}. Desired state: {}. {}/{} instances.",
- currentState,
- runningNodesThreshold == 1.0f ?
- LlapStatusHelpers.State.RUNNING_ALL :
- LlapStatusHelpers.State.RUNNING_PARTIAL,
+ "Waiting until running nodes threshold is reached. Current: {} Desired: {}." +
+ " {}/{} instances.",
+ new DecimalFormat("#.###").format(ratio),
+ new DecimalFormat("#.###")
+ .format(runningNodesThreshold),
statusServiceDriver.appStatusBuilder.getLiveInstances(),
statusServiceDriver.appStatusBuilder
.getDesiredInstances());
}
numAttempts--;
continue;
- }
-
- // we have reached RUNNING state, now check if running nodes threshold is met
- final int liveInstances = statusServiceDriver.appStatusBuilder.getLiveInstances();
- final int desiredInstances = statusServiceDriver.appStatusBuilder.getDesiredInstances();
- if (desiredInstances > 0) {
- final float ratio = (float) liveInstances / (float) desiredInstances;
- if (ratio < runningNodesThreshold) {
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "Waiting until running nodes threshold is reached. Current: {} Desired: {}." +
- " {}/{} instances.",
- new DecimalFormat("#.###").format(ratio),
- new DecimalFormat("#.###")
- .format(runningNodesThreshold),
- statusServiceDriver.appStatusBuilder.getLiveInstances(),
- statusServiceDriver.appStatusBuilder
- .getDesiredInstances());
- }
- numAttempts--;
- continue;
- } else {
- desiredStateAttained = true;
- statusServiceDriver.appStatusBuilder.setRunningThresholdAchieved(true);
- }
} else {
- numAttempts--;
- continue;
+ desiredStateAttained = true;
+ statusServiceDriver.appStatusBuilder.setRunningThresholdAchieved(true);
}
+ } else {
+ numAttempts--;
+ continue;
}
- } else if (ret == ExitCode.YARN_ERROR.getInt() && watchMode) {
- LOG.warn("Watch mode enabled and got YARN error. Retrying..");
- numAttempts--;
- continue;
- } else if (ret == ExitCode.SLIDER_CLIENT_ERROR_CREATE_FAILED.getInt() && watchMode) {
- LOG.warn("Watch mode enabled and slider client creation failed. Retrying..");
- numAttempts--;
- continue;
- } else if (ret == ExitCode.SLIDER_CLIENT_ERROR_OTHER.getInt() && watchMode) {
- LOG.warn("Watch mode enabled and got slider client error. Retrying..");
- numAttempts--;
- continue;
- } else if (ret == ExitCode.LLAP_REGISTRY_ERROR.getInt() && watchMode) {
- LOG.warn("Watch mode enabled and got LLAP registry error. Retrying..");
- numAttempts--;
- continue;
}
- break;
- } finally {
- // TODO Remove this before commit.
+ } else if (ret == ExitCode.YARN_ERROR.getInt() && watchMode) {
+ LOG.warn("Watch mode enabled and got YARN error. Retrying..");
+ numAttempts--;
+ continue;
+ } else if (ret == ExitCode.SLIDER_CLIENT_ERROR_CREATE_FAILED.getInt() && watchMode) {
+ LOG.warn("Watch mode enabled and slider client creation failed. Retrying..");
+ numAttempts--;
+ continue;
+ } else if (ret == ExitCode.SLIDER_CLIENT_ERROR_OTHER.getInt() && watchMode) {
+ LOG.warn("Watch mode enabled and got slider client error. Retrying..");
+ numAttempts--;
+ continue;
+ } else if (ret == ExitCode.LLAP_REGISTRY_ERROR.getInt() && watchMode) {
+ LOG.warn("Watch mode enabled and got LLAP registry error. Retrying..");
+ numAttempts--;
+ continue;
}
+ break;
}
// Log final state to CONSOLE_LOGGER
LlapStatusServiceDriver
@@ -1036,7 +1038,7 @@ public class LlapStatusServiceDriver {
if (lastSummaryLogTime < currentTime - LOG_SUMMARY_INTERVAL) {
String diagString = null;
if (launchingState == null && statusServiceDriver.appStatusBuilder.getState() ==
- LlapStatusHelpers.State.COMPLETE && watchMode) {
+ State.COMPLETE && watchMode) {
// First known state was COMPLETED. Wait for the app launch to start.
diagString = "Awaiting LLAP launch";
// Clear completed instances in this case. Don't want to provide information from the previous run.
[5/6] hive git commit: HIVE-17610 : LLAP IO: an exception in
exception handling can hide the original exception (Sergey Shelukhin,
reviewed by Rajesh Balamohan)
Posted by se...@apache.org.
HIVE-17610 : LLAP IO: an exception in exception handling can hide the original exception (Sergey Shelukhin, reviewed by Rajesh Balamohan)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/1e4097ef
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/1e4097ef
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/1e4097ef
Branch: refs/heads/master
Commit: 1e4097ef767246538c8386f52c49f5207433a98e
Parents: e1e11d4
Author: sergey <se...@apache.org>
Authored: Fri Sep 29 13:41:58 2017 -0700
Committer: sergey <se...@apache.org>
Committed: Fri Sep 29 13:56:18 2017 -0700
----------------------------------------------------------------------
.../hive/llap/io/api/impl/LlapRecordReader.java | 3 +-
.../ql/io/orc/encoded/EncodedReaderImpl.java | 210 +++++++++++--------
2 files changed, 127 insertions(+), 86 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/1e4097ef/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java
index 5d93884..70bd05c 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/LlapRecordReader.java
@@ -379,8 +379,9 @@ class LlapRecordReader
@Override
public void setError(Throwable t) {
counters.incrCounter(LlapIOCounters.NUM_ERRORS);
- LlapIoImpl.LOG.info("setError called; closed {}, done {}, err {}, pending {}",
+ LlapIoImpl.LOG.debug("setError called; current state closed {}, done {}, err {}, pending {}",
isClosed, isDone, pendingError, pendingData.size());
+ LlapIoImpl.LOG.warn("setError called with an error", t);
assert t != null;
synchronized (pendingData) {
pendingError = t;
http://git-wip-us.apache.org/repos/asf/hive/blob/1e4097ef/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
index 467bac2..31d5dd3 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/encoded/EncodedReaderImpl.java
@@ -374,9 +374,10 @@ class EncodedReaderImpl implements EncodedReader {
// CacheChunks, so the list is just CacheChunk-s from that point on.
DiskRangeList iter = preReadUncompressedStreams(stripeOffset, colCtxs, toRead, toRelease);
+ // 4. Finally, decompress data, map per RG, and return to caller.
+ // We go by RG and not by column because that is how data is processed.
+ boolean hasError = true;
try {
- // 4. Finally, decompress data, map per RG, and return to caller.
- // We go by RG and not by column because that is how data is processed.
int rgCount = (int)Math.ceil((double)stripe.getNumberOfRows() / rowIndexStride);
for (int rgIx = 0; rgIx < rgCount; ++rgIx) {
if (rgs != null && !rgs[rgIx]) {
@@ -386,7 +387,7 @@ class EncodedReaderImpl implements EncodedReader {
// Create the batch we will use to return data for this RG.
OrcEncodedColumnBatch ecb = POOLS.ecbPool.take();
trace.logStartRg(rgIx);
- boolean hasError = true;
+ boolean hasErrorForEcb = true;
try {
ecb.init(fileKey, stripeIx, rgIx, included.length);
for (int colIx = 0; colIx < colCtxs.length; ++colIx) {
@@ -462,10 +463,14 @@ class EncodedReaderImpl implements EncodedReader {
}
}
}
- hasError = false;
+ hasErrorForEcb = false;
} finally {
- if (hasError) {
- releaseEcbRefCountsOnError(ecb);
+ if (hasErrorForEcb) {
+ try {
+ releaseEcbRefCountsOnError(ecb);
+ } catch (Throwable t) {
+ LOG.error("Error during the cleanup of an error; ignoring", t);
+ }
}
}
// After this, the non-initial refcounts are the responsibility of the consumer.
@@ -477,35 +482,40 @@ class EncodedReaderImpl implements EncodedReader {
+ RecordReaderUtils.stringifyDiskRanges(toRead.next));
}
trace.logRanges(fileKey, stripeOffset, toRead.next, RangesSrc.PREREAD);
+ hasError = false;
} finally {
- // Release the unreleased stripe-level buffers. See class comment about refcounts.
- for (int colIx = 0; colIx < colCtxs.length; ++colIx) {
- ColumnReadContext ctx = colCtxs[colIx];
- if (ctx == null) continue; // This column is not included.
- for (int streamIx = 0; streamIx < ctx.streamCount; ++streamIx) {
- StreamContext sctx = ctx.streams[streamIx];
- if (sctx == null || sctx.stripeLevelStream == null) continue;
- if (0 != sctx.stripeLevelStream.decRef()) continue;
- // Note - this is a little bit confusing; the special treatment of stripe-level buffers
- // is because we run the ColumnStreamData refcount one ahead (as specified above). It
- // may look like this would release the buffers too many times (one release from the
- // consumer, one from releaseInitialRefcounts below, and one here); however, this is
- // merely handling a special case where all the batches that are sharing the stripe-
- // level stream have been processed before we got here; they have all decRef-ed the CSD,
- // but have not released the buffers because of that extra refCount. So, this is
- // essentially the "consumer" refcount being released here.
- for (MemoryBuffer buf : sctx.stripeLevelStream.getCacheBuffers()) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Unlocking {} at the end of processing", buf);
+ try {
+ // Release the unreleased stripe-level buffers. See class comment about refcounts.
+ for (int colIx = 0; colIx < colCtxs.length; ++colIx) {
+ ColumnReadContext ctx = colCtxs[colIx];
+ if (ctx == null) continue; // This column is not included.
+ for (int streamIx = 0; streamIx < ctx.streamCount; ++streamIx) {
+ StreamContext sctx = ctx.streams[streamIx];
+ if (sctx == null || sctx.stripeLevelStream == null) continue;
+ if (0 != sctx.stripeLevelStream.decRef()) continue;
+ // Note - this is a little bit confusing; the special treatment of stripe-level buffers
+ // is because we run the ColumnStreamData refcount one ahead (as specified above). It
+ // may look like this would release the buffers too many times (one release from the
+ // consumer, one from releaseInitialRefcounts below, and one here); however, this is
+ // merely handling a special case where all the batches that are sharing the stripe-
+ // level stream have been processed before we got here; they have all decRef-ed the CSD,
+ // but have not released the buffers because of that extra refCount. So, this is
+ // essentially the "consumer" refcount being released here.
+ for (MemoryBuffer buf : sctx.stripeLevelStream.getCacheBuffers()) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Unlocking {} at the end of processing", buf);
+ }
+ cacheWrapper.releaseBuffer(buf);
}
- cacheWrapper.releaseBuffer(buf);
}
}
+ releaseInitialRefcounts(toRead.next);
+ // Release buffers as we are done with all the streams... also see toRelease comment.
+ releaseBuffers(toRelease.keySet(), true);
+ } catch (Throwable t) {
+ if (!hasError) throw new IOException(t);
+ LOG.error("Error during the cleanup after another error; ignoring", t);
}
-
- releaseInitialRefcounts(toRead.next);
- // Release buffers as we are done with all the streams... also see toRelease comment.
- releaseBuffers(toRelease.keySet(), true);
}
releaseCacheChunksIntoObjectPool(toRead.next);
}
@@ -623,6 +633,18 @@ class EncodedReaderImpl implements EncodedReader {
while (current != null) {
DiskRangeList toFree = current;
current = current.next;
+ if (toFree instanceof ProcCacheChunk) {
+ ProcCacheChunk pcc = (ProcCacheChunk)toFree;
+ if (pcc.originalData != null) {
+ // This can only happen in case of failure - we read some data, but didn't decompress
+ // it. Deallocate the buffer directly, do not decref.
+ if (pcc.getBuffer() != null) {
+ cacheWrapper.getAllocator().deallocate(pcc.getBuffer());
+ }
+ continue;
+ }
+
+ }
if (!(toFree instanceof CacheChunk)) continue;
CacheChunk cc = (CacheChunk)toFree;
if (cc.getBuffer() == null) continue;
@@ -841,11 +863,16 @@ class EncodedReaderImpl implements EncodedReader {
copyUncompressedChunk(chunk.originalData, dest);
}
- chunk.originalData = null;
if (isTracingEnabled) {
LOG.trace("Locking " + chunk.getBuffer() + " due to reuse (after decompression)");
}
- cacheWrapper.reuseBuffer(chunk.getBuffer());
+ // After we set originalData to null, we incref the buffer and the cleanup would decref it.
+ // Note that this assumes the failure during incref means incref didn't occur.
+ try {
+ cacheWrapper.reuseBuffer(chunk.getBuffer());
+ } finally {
+ chunk.originalData = null;
+ }
}
// 5. Release the copies we made directly to the cleaner.
@@ -1847,60 +1874,69 @@ class EncodedReaderImpl implements EncodedReader {
DiskRangeList iter = preReadUncompressedStreams(stripeOffset, colCtxs, toRead, toRelease);
// 4. Decompress the data.
- for (int colIx = 0; colIx < colCtxs.length; ++colIx) {
- ReadContext ctx = colCtxs[colIx];
- if (ctx == null) continue; // This column is not included.
- for (int streamIx = 0; streamIx < ctx.streamCount; ++streamIx) {
- StreamContext sctx = ctx.streams[streamIx];
- try {
- if (isTracingEnabled) {
- LOG.trace("Getting index stream " + sctx.kind + " for column " + ctx.colIx
- + " at " + sctx.offset + ", " + sctx.length);
- }
- ColumnStreamData csd = POOLS.csdPool.take();
- long endCOffset = sctx.offset + sctx.length;
- DiskRangeList lastCached = readEncodedStream(stripeOffset, iter, sctx.offset,
- endCOffset, csd, endCOffset, sctx.offset, toRelease);
- if (lastCached != null) {
- iter = lastCached;
- }
- CodedInputStream cis = CodedInputStream.newInstance(
- new IndexStream(csd.getCacheBuffers(), sctx.length));
- cis.setSizeLimit(InStream.PROTOBUF_MESSAGE_MAX_LIMIT);
- switch (sctx.kind) {
- case ROW_INDEX:
- index.getRowGroupIndex()[colIx] = OrcProto.RowIndex.parseFrom(cis);
- break;
- case BLOOM_FILTER:
- case BLOOM_FILTER_UTF8:
- index.getBloomFilterIndex()[colIx] = OrcProto.BloomFilterIndex.parseFrom(cis);
- break;
- default:
- throw new AssertionError("Unexpected index stream type " + sctx.kind);
- }
- // We are done with the buffers; unlike data blocks, we are also the consumer. Release.
- for (MemoryBuffer buf : csd.getCacheBuffers()) {
- if (buf == null) continue;
- cacheWrapper.releaseBuffer(buf);
+ boolean hasError = true;
+ try {
+ for (int colIx = 0; colIx < colCtxs.length; ++colIx) {
+ ReadContext ctx = colCtxs[colIx];
+ if (ctx == null) continue; // This column is not included.
+ for (int streamIx = 0; streamIx < ctx.streamCount; ++streamIx) {
+ StreamContext sctx = ctx.streams[streamIx];
+ try {
+ if (isTracingEnabled) {
+ LOG.trace("Getting index stream " + sctx.kind + " for column " + ctx.colIx
+ + " at " + sctx.offset + ", " + sctx.length);
+ }
+ ColumnStreamData csd = POOLS.csdPool.take();
+ long endCOffset = sctx.offset + sctx.length;
+ DiskRangeList lastCached = readEncodedStream(stripeOffset, iter, sctx.offset,
+ endCOffset, csd, endCOffset, sctx.offset, toRelease);
+ if (lastCached != null) {
+ iter = lastCached;
+ }
+ CodedInputStream cis = CodedInputStream.newInstance(
+ new IndexStream(csd.getCacheBuffers(), sctx.length));
+ cis.setSizeLimit(InStream.PROTOBUF_MESSAGE_MAX_LIMIT);
+ switch (sctx.kind) {
+ case ROW_INDEX:
+ index.getRowGroupIndex()[colIx] = OrcProto.RowIndex.parseFrom(cis);
+ break;
+ case BLOOM_FILTER:
+ case BLOOM_FILTER_UTF8:
+ index.getBloomFilterIndex()[colIx] = OrcProto.BloomFilterIndex.parseFrom(cis);
+ break;
+ default:
+ throw new AssertionError("Unexpected index stream type " + sctx.kind);
+ }
+ // We are done with the buffers; unlike data blocks, we are also the consumer. Release.
+ for (MemoryBuffer buf : csd.getCacheBuffers()) {
+ if (buf == null) continue;
+ cacheWrapper.releaseBuffer(buf);
+ }
+ } catch (Exception ex) {
+ DiskRangeList drl = toRead == null ? null : toRead.next;
+ LOG.error("Error getting stream " + sctx.kind + " for column " + ctx.colIx
+ + " at " + sctx.offset + ", " + sctx.length + "; toRead "
+ + RecordReaderUtils.stringifyDiskRanges(drl), ex);
+ throw (ex instanceof IOException) ? (IOException)ex : new IOException(ex);
}
- } catch (Exception ex) {
- DiskRangeList drl = toRead == null ? null : toRead.next;
- LOG.error("Error getting stream " + sctx.kind + " for column " + ctx.colIx
- + " at " + sctx.offset + ", " + sctx.length + "; toRead "
- + RecordReaderUtils.stringifyDiskRanges(drl), ex);
- throw (ex instanceof IOException) ? (IOException)ex : new IOException(ex);
}
}
- }
- if (isTracingEnabled) {
- LOG.trace("Disk ranges after preparing all the data "
- + RecordReaderUtils.stringifyDiskRanges(toRead.next));
+ if (isTracingEnabled) {
+ LOG.trace("Disk ranges after preparing all the data "
+ + RecordReaderUtils.stringifyDiskRanges(toRead.next));
+ }
+ hasError = false;
+ } finally {
+ // Release the unreleased buffers. See class comment about refcounts.
+ try {
+ releaseInitialRefcounts(toRead.next);
+ releaseBuffers(toRelease.keySet(), true);
+ } catch (Throwable t) {
+ if (!hasError) throw new IOException(t);
+ LOG.error("Error during the cleanup after another error; ignoring", t);
+ }
}
-
- // Release the unreleased buffers. See class comment about refcounts.
- releaseInitialRefcounts(toRead.next);
- releaseBuffers(toRelease.keySet(), true);
releaseCacheChunksIntoObjectPool(toRead.next);
}
@@ -1939,10 +1975,14 @@ class EncodedReaderImpl implements EncodedReader {
// At this point, everything in the list is going to have a refcount of one. Unless it
// failed between the allocation and the incref for a single item, we should be ok.
if (hasError) {
- releaseInitialRefcounts(toRead.next);
- if (toRelease != null) {
- releaseBuffers(toRelease.keySet(), true);
- toRelease.clear();
+ try {
+ releaseInitialRefcounts(toRead.next);
+ if (toRelease != null) {
+ releaseBuffers(toRelease.keySet(), true);
+ toRelease.clear();
+ }
+ } catch (Throwable t) {
+ LOG.error("Error during the cleanup after another error; ignoring", t);
}
}
}
[2/6] hive git commit: HIVE-17639 : don't reuse planner context when
re-parsing the query (Sergey Shelukhin, reviewed by Ashutosh Chauhan)
Posted by se...@apache.org.
HIVE-17639 : don't reuse planner context when re-parsing the query (Sergey Shelukhin, reviewed by Ashutosh Chauhan)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/e1e11d4a
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/e1e11d4a
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/e1e11d4a
Branch: refs/heads/master
Commit: e1e11d4a85a1a8d0b12494e2c69600d4cef1dcaa
Parents: 32b9716
Author: sergey <se...@apache.org>
Authored: Fri Sep 29 13:36:57 2017 -0700
Committer: sergey <se...@apache.org>
Committed: Fri Sep 29 13:56:18 2017 -0700
----------------------------------------------------------------------
.../apache/hadoop/hive/ql/parse/CalcitePlanner.java | 10 ++++++++--
.../hadoop/hive/ql/parse/SemanticAnalyzer.java | 15 +++++++++++++--
2 files changed, 21 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/e1e11d4a/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java
index 28953b9..e3ae296 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java
@@ -228,6 +228,8 @@ import org.apache.hadoop.hive.ql.parse.PTFInvocationSpec.OrderSpec;
import org.apache.hadoop.hive.ql.parse.PTFInvocationSpec.PartitionExpression;
import org.apache.hadoop.hive.ql.parse.PTFInvocationSpec.PartitionSpec;
import org.apache.hadoop.hive.ql.parse.QBExpr.Opcode;
+import org.apache.hadoop.hive.ql.parse.SemanticAnalyzer.PlannerContext;
+import org.apache.hadoop.hive.ql.parse.SemanticAnalyzer.PlannerContextFactory;
import org.apache.hadoop.hive.ql.parse.WindowingSpec.BoundarySpec;
import org.apache.hadoop.hive.ql.parse.WindowingSpec.WindowExpressionSpec;
import org.apache.hadoop.hive.ql.parse.WindowingSpec.WindowFunctionSpec;
@@ -290,8 +292,12 @@ public class CalcitePlanner extends SemanticAnalyzer {
@SuppressWarnings("nls")
public void analyzeInternal(ASTNode ast) throws SemanticException {
if (runCBO) {
- PreCboCtx cboCtx = new PreCboCtx();
- super.analyzeInternal(ast, cboCtx);
+ super.analyzeInternal(ast, new PlannerContextFactory() {
+ @Override
+ public PlannerContext create() {
+ return new PreCboCtx();
+ }
+ });
} else {
super.analyzeInternal(ast);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/e1e11d4a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
index 68ddcf0..2fbdd70 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
@@ -10971,7 +10971,16 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
@Override
@SuppressWarnings("nls")
public void analyzeInternal(ASTNode ast) throws SemanticException {
- analyzeInternal(ast, new PlannerContext());
+ analyzeInternal(ast, new PlannerContextFactory() {
+ @Override
+ public PlannerContext create() {
+ return new PlannerContext();
+ }
+ });
+ }
+
+ protected static interface PlannerContextFactory {
+ PlannerContext create();
}
/**
@@ -11365,11 +11374,12 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
}
}
- void analyzeInternal(ASTNode ast, PlannerContext plannerCtx) throws SemanticException {
+ void analyzeInternal(ASTNode ast, PlannerContextFactory pcf) throws SemanticException {
// 1. Generate Resolved Parse tree from syntax tree
LOG.info("Starting Semantic Analysis");
//change the location of position alias process here
processPositionAlias(ast);
+ PlannerContext plannerCtx = pcf.create();
if (!genResolvedParseTree(ast, plannerCtx)) {
return;
}
@@ -11388,6 +11398,7 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer {
// Here we rewrite the * and also the masking table
ASTNode tree = rewriteASTWithMaskAndFilter(ast);
if (tree != ast) {
+ plannerCtx = pcf.create();
ctx.setSkipTableMasking(true);
init(true);
//change the location of position alias process here
[3/6] hive git commit: HIVE-17644 : directSQL errors out on key
constraints until the DB is initialized (Sergey Shelukhin,
reviewed by Ashutosh Chauhan)
Posted by se...@apache.org.
HIVE-17644 : directSQL errors out on key constraints until the DB is initialized (Sergey Shelukhin, reviewed by Ashutosh Chauhan)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/3a5565e2
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/3a5565e2
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/3a5565e2
Branch: refs/heads/master
Commit: 3a5565e2987d59fdd41ca67a1a6f4b893b03e5e2
Parents: b3dbe31
Author: sergey <se...@apache.org>
Authored: Fri Sep 29 13:30:34 2017 -0700
Committer: sergey <se...@apache.org>
Committed: Fri Sep 29 13:56:18 2017 -0700
----------------------------------------------------------------------
.../hive/metastore/MetaStoreDirectSql.java | 48 ++++++++------------
1 file changed, 19 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/3a5565e2/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
----------------------------------------------------------------------
diff --git a/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java b/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
index a940040..6ae6f32 100644
--- a/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
+++ b/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
@@ -32,6 +32,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
@@ -211,29 +212,21 @@ class MetaStoreDirectSql {
tx.begin();
doCommit = true;
}
- Query dbQuery = null, tblColumnQuery = null, partColumnQuery = null;
-
+ LinkedList<Query> initQueries = new LinkedList<>();
+
try {
// Force the underlying db to initialize.
- dbQuery = pm.newQuery(MDatabase.class, "name == ''");
- dbQuery.execute();
-
- tblColumnQuery = pm.newQuery(MTableColumnStatistics.class, "dbName == ''");
- tblColumnQuery.execute();
-
- partColumnQuery = pm.newQuery(MPartitionColumnStatistics.class, "dbName == ''");
- partColumnQuery.execute();
-
- /*
- these queries for the notification related tables have to be executed so
- that the tables are created. This was not required earlier because we were
- interacting with these tables via DataNucleus so it would create them if
- they did not exist (mostly used in test, schematool should be used for production).
- however this has been changed and we used direct SQL
- queries via DataNucleus to interact with them now.
- */
- pm.newQuery(MNotificationLog.class, "dbName == ''").execute();
- pm.newQuery(MNotificationNextId.class, "nextEventId < -1").execute();
+ initQueries.add(pm.newQuery(MDatabase.class, "name == ''"));
+ initQueries.add(pm.newQuery(MTableColumnStatistics.class, "dbName == ''"));
+ initQueries.add(pm.newQuery(MPartitionColumnStatistics.class, "dbName == ''"));
+ initQueries.add(pm.newQuery(MConstraint.class, "childIntegerIndex < 0"));
+ initQueries.add(pm.newQuery(MNotificationLog.class, "dbName == ''"));
+ initQueries.add(pm.newQuery(MNotificationNextId.class, "nextEventId < -1"));
+ Query q;
+ while ((q = initQueries.peekFirst()) != null) {
+ q.execute();
+ initQueries.pollFirst();
+ }
return true;
} catch (Exception ex) {
@@ -245,14 +238,11 @@ class MetaStoreDirectSql {
if (doCommit) {
tx.commit();
}
- if (dbQuery != null) {
- dbQuery.closeAll();
- }
- if (tblColumnQuery != null) {
- tblColumnQuery.closeAll();
- }
- if (partColumnQuery != null) {
- partColumnQuery.closeAll();
+ for (Query q : initQueries) {
+ try {
+ q.closeAll();
+ } catch (Throwable t) {
+ }
}
}
}
[6/6] hive git commit: HIVE-17628 : always use fully qualified path
for tables/partitions/etc. (Sergey Shelukhin,
reviewed by Ashutosh Chauhan and Thejas M Nair)
Posted by se...@apache.org.
HIVE-17628 : always use fully qualified path for tables/partitions/etc. (Sergey Shelukhin, reviewed by Ashutosh Chauhan and Thejas M Nair)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/32b9716f
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/32b9716f
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/32b9716f
Branch: refs/heads/master
Commit: 32b9716fe73b89198ef2714039127e451669e5fb
Parents: a860795
Author: sergey <se...@apache.org>
Authored: Fri Sep 29 13:34:15 2017 -0700
Committer: sergey <se...@apache.org>
Committed: Fri Sep 29 13:56:18 2017 -0700
----------------------------------------------------------------------
.../org/apache/hadoop/hive/ql/exec/DDLTask.java | 18 ++++++++----------
.../org/apache/hadoop/hive/ql/exec/Utilities.java | 10 ----------
.../org/apache/hadoop/hive/ql/metadata/Hive.java | 6 +++---
.../hadoop/hive/ql/plan/CreateTableDesc.java | 2 +-
.../test/results/clientnegative/external2.q.out | 2 +-
5 files changed, 13 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/32b9716f/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java
index 2cf5bfd..2e4e2fa 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java
@@ -1016,10 +1016,8 @@ public class DDLTask extends Task<DDLWork> implements Serializable {
}
String indexTableName = crtIndex.getIndexTableName();
- if (!Utilities.isDefaultNameNode(conf)) {
- // If location is specified - ensure that it is a full qualified name
- makeLocationQualified(crtIndex, indexTableName);
- }
+ // If location is specified - ensure that it is a full qualified name
+ makeLocationQualified(crtIndex, indexTableName);
db
.createIndex(
@@ -4363,9 +4361,7 @@ public class DDLTask extends Task<DDLWork> implements Serializable {
database.setOwnerName(SessionState.getUserFromAuthenticator());
database.setOwnerType(PrincipalType.USER);
try {
- if (!Utilities.isDefaultNameNode(conf)) {
- makeLocationQualified(database);
- }
+ makeLocationQualified(database);
db.createDatabase(database, crtDb.getIfNotExists());
}
catch (AlreadyExistsException ex) {
@@ -4631,8 +4627,8 @@ public class DDLTask extends Task<DDLWork> implements Serializable {
}
}
- if (!Utilities.isDefaultNameNode(conf)) {
- // If location is specified - ensure that it is a full qualified name
+ // If location is specified - ensure that it is a full qualified name
+ if (DDLTask.doesTableNeedLocation(tbl)) {
makeLocationQualified(tbl.getDbName(), tbl.getTTable().getSd(), tbl.getTableName(), conf);
}
@@ -4954,9 +4950,11 @@ public class DDLTask extends Task<DDLWork> implements Serializable {
// this method could be moved to the HiveStorageHandler interface.
boolean retval = true;
if (tbl.getStorageHandler() != null) {
+ // TODO: why doesn't this check class name rather than toString?
String sh = tbl.getStorageHandler().toString();
retval = !sh.equals("org.apache.hadoop.hive.hbase.HBaseStorageHandler")
- && !sh.equals(Constants.DRUID_HIVE_STORAGE_HANDLER_ID);
+ && !sh.equals(Constants.DRUID_HIVE_STORAGE_HANDLER_ID)
+ && !sh.equals("org.apache.hadoop.hive.accumulo.AccumuloStorageHandler");
}
return retval;
}
http://git-wip-us.apache.org/repos/asf/hive/blob/32b9716f/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
index ae70cba..ae63727 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
@@ -3666,16 +3666,6 @@ public final class Utilities {
}
/**
- * Checks if current hive script was executed with non-default namenode
- *
- * @return True/False
- */
- public static boolean isDefaultNameNode(HiveConf conf) {
- return !conf.getChangedProperties().containsKey(
- CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY);
- }
-
- /**
* Checks if the current HiveServer2 logging operation level is >= PERFORMANCE.
* @param conf Hive configuration.
* @return true if current HiveServer2 logging operation level is >= PERFORMANCE.
http://git-wip-us.apache.org/repos/asf/hive/blob/32b9716f/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
index e8a0474..436a2fe 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
@@ -703,7 +703,7 @@ public class Hive {
try {
validatePartition(newPart);
String location = newPart.getLocation();
- if (location != null && !Utilities.isDefaultNameNode(conf)) {
+ if (location != null) {
location = Utilities.getQualifiedPath(conf, new Path(location));
newPart.setLocation(location);
}
@@ -747,7 +747,7 @@ public class Hive {
tmpPart.getParameters().remove(hive_metastoreConstants.DDL_TIME);
}
String location = tmpPart.getLocation();
- if (location != null && !Utilities.isDefaultNameNode(conf)) {
+ if (location != null) {
location = Utilities.getQualifiedPath(conf, new Path(location));
tmpPart.setLocation(location);
}
@@ -2235,7 +2235,7 @@ private void constructOneLBLocationMap(FileStatus fSta,
Table tbl, AddPartitionDesc.OnePartitionDesc addSpec) throws HiveException {
Path location = addSpec.getLocation() != null
? new Path(tbl.getPath(), addSpec.getLocation()) : null;
- if (location !=null && !Utilities.isDefaultNameNode(conf)) {
+ if (location != null) {
// Ensure that it is a full qualified path (in most cases it will be since tbl.getPath() is full qualified)
location = new Path(Utilities.getQualifiedPath(conf, location));
}
http://git-wip-us.apache.org/repos/asf/hive/blob/32b9716f/ql/src/java/org/apache/hadoop/hive/ql/plan/CreateTableDesc.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/CreateTableDesc.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/CreateTableDesc.java
index 4320421..a05cb96 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/plan/CreateTableDesc.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/CreateTableDesc.java
@@ -804,7 +804,7 @@ public class CreateTableDesc extends DDLDesc implements Serializable {
tbl.getTTable().getSd().setOutputFormat(tbl.getOutputFormatClass().getName());
}
- if (!Utilities.isDefaultNameNode(conf) && DDLTask.doesTableNeedLocation(tbl)) {
+ if (DDLTask.doesTableNeedLocation(tbl)) {
// If location is specified - ensure that it is a full qualified name
DDLTask.makeLocationQualified(tbl.getDbName(), tbl.getTTable().getSd(), tableName, conf);
}
http://git-wip-us.apache.org/repos/asf/hive/blob/32b9716f/ql/src/test/results/clientnegative/external2.q.out
----------------------------------------------------------------------
diff --git a/ql/src/test/results/clientnegative/external2.q.out b/ql/src/test/results/clientnegative/external2.q.out
index 706432a..eb5518c 100644
--- a/ql/src/test/results/clientnegative/external2.q.out
+++ b/ql/src/test/results/clientnegative/external2.q.out
@@ -10,4 +10,4 @@ POSTHOOK: Output: default@external2
PREHOOK: type: ALTERTABLE_ADDPARTS
#### A masked pattern was here ####
PREHOOK: Output: default@external2
-FAILED: Execution Error, return code 1 from org.apache.hadoop.hive.ql.exec.DDLTask. MetaException(message:java.io.IOException: Got exception: java.io.IOException No FileSystem for scheme: invalidscheme)
+FAILED: Execution Error, return code 1 from org.apache.hadoop.hive.ql.exec.DDLTask. java.io.IOException: No FileSystem for scheme: invalidscheme
[4/6] hive git commit: HIVE-17643 : recent WM changes broke reopen
due to spurious overloads (Sergey Shelukhin,
reviewed by Prasanth Jayachandran)
Posted by se...@apache.org.
HIVE-17643 : recent WM changes broke reopen due to spurious overloads (Sergey Shelukhin, reviewed by Prasanth Jayachandran)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/a860795c
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/a860795c
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/a860795c
Branch: refs/heads/master
Commit: a860795ce011f024b7b66bba4a000a28eb9fbc9d
Parents: 3a5565e
Author: sergey <se...@apache.org>
Authored: Fri Sep 29 13:31:40 2017 -0700
Committer: sergey <se...@apache.org>
Committed: Fri Sep 29 13:56:18 2017 -0700
----------------------------------------------------------------------
.../hive/ql/exec/tez/TezSessionPoolManager.java | 19 ++++---------------
.../hive/ql/exec/tez/TezSessionPoolSession.java | 4 ++--
.../hadoop/hive/ql/exec/tez/TezSessionState.java | 4 ++--
.../apache/hadoop/hive/ql/exec/tez/TezTask.java | 3 +++
.../hadoop/hive/ql/exec/tez/WorkloadManager.java | 11 ++++-------
.../hive/ql/exec/tez/TestTezSessionPool.java | 16 ++++++++--------
6 files changed, 23 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/a860795c/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java
index 538d745..edcecb0 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolManager.java
@@ -18,8 +18,6 @@
package org.apache.hadoop.hive.ql.exec.tez;
-import org.apache.hadoop.hive.registry.impl.TezAmInstance;
-
import java.util.HashSet;
import java.util.concurrent.Semaphore;
@@ -321,7 +319,8 @@ public class TezSessionPoolManager
* the session to be closed
* @throws Exception
*/
- public void destroySession(TezSessionState tezSessionState) throws Exception {
+ @Override
+ public void destroy(TezSessionState tezSessionState) throws Exception {
LOG.warn("We are closing a " + (tezSessionState.isDefault() ? "default" : "non-default")
+ " session because of retry failure.");
tezSessionState.close(false);
@@ -398,7 +397,8 @@ public class TezSessionPoolManager
}
/** Reopens the session that was found to not be running. */
- public TezSessionState reopenSession(TezSessionState sessionState,
+ @Override
+ public TezSessionState reopen(TezSessionState sessionState,
Configuration conf, String[] additionalFiles) throws Exception {
HiveConf sessionConf = sessionState.getConf();
if (sessionState.getQueueName() != null
@@ -467,15 +467,4 @@ public class TezSessionPoolManager
public SessionExpirationTracker getExpirationTracker() {
return expirationTracker;
}
-
- @Override
- public TezSessionPoolSession reopen(
- TezSessionPoolSession session, Configuration conf, String[] inputOutputJars) {
- return reopen(session, conf, inputOutputJars);
- }
-
- @Override
- public void destroy(TezSessionPoolSession session) throws Exception {
- destroySession(session);
- }
}
http://git-wip-us.apache.org/repos/asf/hive/blob/a860795c/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolSession.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolSession.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolSession.java
index 4488c12..694f15b 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolSession.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionPoolSession.java
@@ -56,9 +56,9 @@ class TezSessionPoolSession extends TezSessionState {
void registerOpenSession(TezSessionPoolSession session);
void unregisterOpenSession(TezSessionPoolSession session);
void returnAfterUse(TezSessionPoolSession session) throws Exception;
- TezSessionState reopen(TezSessionPoolSession session, Configuration conf,
+ TezSessionState reopen(TezSessionState session, Configuration conf,
String[] inputOutputJars) throws Exception;
- void destroy(TezSessionPoolSession session) throws Exception;
+ void destroy(TezSessionState session) throws Exception;
}
private final AtomicInteger sessionState = new AtomicInteger(STATE_NONE);
http://git-wip-us.apache.org/repos/asf/hive/blob/a860795c/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java
index e5850f9..363443d 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezSessionState.java
@@ -744,11 +744,11 @@ public class TezSessionState {
public TezSessionState reopen(
Configuration conf, String[] inputOutputJars) throws Exception {
// By default, TezSessionPoolManager handles this for both pool and non-pool session.
- return TezSessionPoolManager.getInstance().reopenSession(this, conf, inputOutputJars);
+ return TezSessionPoolManager.getInstance().reopen(this, conf, inputOutputJars);
}
public void destroy() throws Exception {
// By default, TezSessionPoolManager handles this for both pool and non-pool session.
- TezSessionPoolManager.getInstance().destroySession(this);
+ TezSessionPoolManager.getInstance().destroy(this);
}
}
http://git-wip-us.apache.org/repos/asf/hive/blob/a860795c/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
index 29d6fe6..28d91cc 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java
@@ -236,6 +236,9 @@ public class TezTask extends Task<TezWork> {
counters = null;
}
} finally {
+ // Note: due to TEZ-3846, the session may actually be invalid in case of some errors.
+ // Currently, reopen on an attempted reuse will take care of that; we cannot tell
+ // if the session is usable until we try.
// We return this to the pool even if it's unusable; reopen is supposed to handle this.
try {
session.returnToSessionManager();
http://git-wip-us.apache.org/repos/asf/hive/blob/a860795c/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/WorkloadManager.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/WorkloadManager.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/WorkloadManager.java
index 288d705..3f62127 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/WorkloadManager.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/WorkloadManager.java
@@ -21,15 +21,12 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.TimeUnit;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.IdentityHashMap;
import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
@@ -308,8 +305,8 @@ public class WorkloadManager
sessions.replaceSession(ensureOwnedSession(oldSession), createSession(), false, null, null);
}
- private WmTezSession ensureOwnedSession(TezSessionPoolSession oldSession) {
- if (!(oldSession instanceof WmTezSession) || !oldSession.isOwnedBy(this)) {
+ private WmTezSession ensureOwnedSession(TezSessionState oldSession) {
+ if (!(oldSession instanceof WmTezSession) || !((WmTezSession)oldSession).isOwnedBy(this)) {
throw new AssertionError("Not a WM session " + oldSession);
}
WmTezSession session = (WmTezSession) oldSession;
@@ -338,7 +335,7 @@ public class WorkloadManager
}
@Override
- public TezSessionState reopen(TezSessionPoolSession session, Configuration conf,
+ public TezSessionState reopen(TezSessionState session, Configuration conf,
String[] additionalFiles) throws Exception {
WmTezSession oldSession = ensureOwnedSession(session), newSession = createSession();
newSession.setPoolName(oldSession.getPoolName());
@@ -358,7 +355,7 @@ public class WorkloadManager
}
@Override
- public void destroy(TezSessionPoolSession session) throws Exception {
+ public void destroy(TezSessionState session) throws Exception {
LOG.warn("Closing a pool session because of retry failure.");
// We never want to lose pool sessions. Replace it instead; al trigger duck redistribution.
WmTezSession wmSession = ensureOwnedSession(session);
http://git-wip-us.apache.org/repos/asf/hive/blob/a860795c/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestTezSessionPool.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestTezSessionPool.java b/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestTezSessionPool.java
index b9f9f5e..05eb761 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestTezSessionPool.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/exec/tez/TestTezSessionPool.java
@@ -190,7 +190,7 @@ public class TestTezSessionPool {
Mockito.when(session.isDefault()).thenReturn(false);
Mockito.when(session.getConf()).thenReturn(conf);
- poolManager.reopenSession(session, conf, null);
+ poolManager.reopen(session, conf, null);
Mockito.verify(session).close(true);
Mockito.verify(session).open(new HashSet<String>(), null);
@@ -200,12 +200,12 @@ public class TestTezSessionPool {
// user explicitly specified queue name
conf.set("tez.queue.name", "tezq1");
- poolManager.reopenSession(session, conf, null);
+ poolManager.reopen(session, conf, null);
assertEquals("tezq1", poolManager.getSession(null, conf, false, false).getQueueName());
// user unsets queue name, will fallback to default session queue
conf.unset("tez.queue.name");
- poolManager.reopenSession(session, conf, null);
+ poolManager.reopen(session, conf, null);
assertEquals("default", poolManager.getSession(null, conf, false, false).getQueueName());
// session.open will unset the queue name from conf but Mockito intercepts the open call
@@ -213,17 +213,17 @@ public class TestTezSessionPool {
conf.unset("tez.queue.name");
// change session's default queue to tezq1 and rerun test sequence
Mockito.when(session.getQueueName()).thenReturn("tezq1");
- poolManager.reopenSession(session, conf, null);
+ poolManager.reopen(session, conf, null);
assertEquals("tezq1", poolManager.getSession(null, conf, false, false).getQueueName());
// user sets default queue now
conf.set("tez.queue.name", "default");
- poolManager.reopenSession(session, conf, null);
+ poolManager.reopen(session, conf, null);
assertEquals("default", poolManager.getSession(null, conf, false, false).getQueueName());
// user does not specify queue so use session default
conf.unset("tez.queue.name");
- poolManager.reopenSession(session, conf, null);
+ poolManager.reopen(session, conf, null);
assertEquals("tezq1", poolManager.getSession(null, conf, false, false).getQueueName());
} catch (Exception e) {
e.printStackTrace();
@@ -328,7 +328,7 @@ public class TestTezSessionPool {
Mockito.when(session.isDefault()).thenReturn(false);
Mockito.when(session.getConf()).thenReturn(conf);
- poolManager.reopenSession(session, conf, null);
+ poolManager.reopen(session, conf, null);
Mockito.verify(session).close(true);
Mockito.verify(session).open(new HashSet<String>(), null);
@@ -340,6 +340,6 @@ public class TestTezSessionPool {
TezSessionState session = Mockito.mock(TezSessionState.class);
Mockito.when(session.isDefault()).thenReturn(false);
- poolManager.destroySession(session);
+ poolManager.destroy(session);
}
}