SLIDER-276 : AgentProvider releases nodes that the AM has already been released after detecting heartbeat failure
Project: http://git-wip-us.apache.org/repos/asf/incubator-slider/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-slider/commit/32fd3f56 Tree: http://git-wip-us.apache.org/repos/asf/incubator-slider/tree/32fd3f56 Diff: http://git-wip-us.apache.org/repos/asf/incubator-slider/diff/32fd3f56 Branch: refs/heads/feature/SLIDER-151_REST_API Commit: 32fd3f56a785e29b55c0560c6256e50bf07215b8 Parents: d68be5a Author: Steve Loughran <ste...@apache.org> Authored: Wed Aug 6 14:41:00 2014 +0100 Committer: Steve Loughran <ste...@apache.org> Committed: Wed Aug 6 14:41:00 2014 +0100 ---------------------------------------------------------------------- .../providers/agent/AgentProviderService.java | 34 ++--- .../providers/agent/ComponentCommandOrder.java | 8 +- .../providers/agent/ComponentInstanceState.java | 80 +++++++--- .../providers/agent/HeartbeatMonitor.java | 66 +++++--- .../server/appmaster/AMViewForProviders.java | 21 ++- .../server/appmaster/SliderAppMaster.java | 27 ++-- .../slider/server/appmaster/state/AppState.java | 6 +- .../appmaster/model/mock/MockContainer.groovy | 2 +- .../agent/TestAgentProviderService.java | 151 +++---------------- .../agent/TestComponentCommandOrder.java | 17 ++- .../agent/TestComponentInstanceState.java | 124 ++++++++------- .../providers/agent/TestHeartbeatMonitor.java | 120 ++++++++++----- 12 files changed, 337 insertions(+), 319 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java b/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java index 6cfadf3..4d54509 100644 --- a/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java +++ b/slider-core/src/main/java/org/apache/slider/providers/agent/AgentProviderService.java @@ -58,6 +58,7 @@ import org.apache.slider.providers.agent.application.metadata.ExportGroup; import org.apache.slider.providers.agent.application.metadata.Metainfo; import org.apache.slider.providers.agent.application.metadata.OSPackage; import org.apache.slider.providers.agent.application.metadata.OSSpecific; +import org.apache.slider.server.appmaster.AMViewForProviders; import org.apache.slider.server.appmaster.state.StateAccessForProviders; import org.apache.slider.server.appmaster.web.rest.agent.AgentCommandType; import org.apache.slider.server.appmaster.web.rest.agent.AgentRestOperations; @@ -127,10 +128,10 @@ public class AgentProviderService extends AbstractProviderService implements private Boolean canAnyMasterPublish = null; private AgentLaunchParameter agentLaunchParameter = null; - private Map<String, ComponentInstanceState> componentStatuses = new ConcurrentHashMap<>(); - private Map<String, Map<String, String>> componentInstanceData = new ConcurrentHashMap(); + private final Map<String, ComponentInstanceState> componentStatuses = new ConcurrentHashMap<>(); + private final Map<String, Map<String, String>> componentInstanceData = new ConcurrentHashMap<>(); private final Map<String, Map<String, String>> allocatedPorts = new ConcurrentHashMap<>(); - private Map<String, String> workFolders = + private final Map<String, String> workFolders = Collections.synchronizedMap(new LinkedHashMap<String, String>(MAX_LOG_ENTRIES, 0.75f, false) { protected boolean removeEldestEntry(Map.Entry eldest) { return size() > MAX_LOG_ENTRIES; @@ -291,7 +292,7 @@ public class AgentProviderService extends AbstractProviderService implements getComponentStatuses().put(label, new ComponentInstanceState( role, - container.getId().toString(), + container.getId(), getClusterInfoPropertyValue(OptionKeys.APPLICATION_NAME))); } @@ -333,7 +334,7 @@ public class AgentProviderService extends AbstractProviderService implements String label = registration.getHostname(); if (getComponentStatuses().containsKey(label)) { response.setResponseStatus(RegistrationStatus.OK); - getComponentStatuses().get(label).setLastHeartbeat(System.currentTimeMillis()); + getComponentStatuses().get(label).heartbeat(System.currentTimeMillis()); } else { response.setResponseStatus(RegistrationStatus.FAILED); response.setLog("Label not recognized."); @@ -371,7 +372,7 @@ public class AgentProviderService extends AbstractProviderService implements Boolean isMaster = isMaster(roleName); ComponentInstanceState componentStatus = getComponentStatuses().get(label); - componentStatus.setLastHeartbeat(System.currentTimeMillis()); + componentStatus.heartbeat(System.currentTimeMillis()); // If no Master can explicitly publish then publish if its a master // Otherwise, wait till the master that can publish is ready if (isMaster && @@ -591,20 +592,18 @@ public class AgentProviderService extends AbstractProviderService implements /** * Lost heartbeat from the container - release it and ask for a replacement * + * * @param label + * @param containerId * - * @return if release is requested successfully + * @return outcome of the operation */ - protected boolean releaseContainer(String label) { + protected AMViewForProviders.ContainerLossReportOutcomes lostContainer( + String label, + ContainerId containerId) throws + SliderException { getComponentStatuses().remove(label); - try { - getAppMaster().refreshContainer(getContainerId(label), true); - } catch (SliderException e) { - log.info("Error while requesting container release for {}. Message: {}", label, e.getMessage()); - return false; - } - - return true; + return getAppMaster().providerLostContainer(containerId); } /** @@ -1059,7 +1058,8 @@ public class AgentProviderService extends AbstractProviderService implements if (!this.allocatedPorts.containsKey(containerId)) { synchronized (this.allocatedPorts) { if (!this.allocatedPorts.containsKey(containerId)) { - this.allocatedPorts.put(containerId, new ConcurrentHashMap()); + this.allocatedPorts.put(containerId, + new ConcurrentHashMap<String, String>()); } } } http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/main/java/org/apache/slider/providers/agent/ComponentCommandOrder.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/providers/agent/ComponentCommandOrder.java b/slider-core/src/main/java/org/apache/slider/providers/agent/ComponentCommandOrder.java index 0dce4bb..a23cc38 100644 --- a/slider-core/src/main/java/org/apache/slider/providers/agent/ComponentCommandOrder.java +++ b/slider-core/src/main/java/org/apache/slider/providers/agent/ComponentCommandOrder.java @@ -126,18 +126,18 @@ public class ComponentCommandOrder { for (ComponentState stateToMatch : required) { for (ComponentInstanceState currState : currentStates) { log.debug("Checking schedule {} {} against dependency {} is {}", - component, command, currState.getCompName(), currState.getState()); - if (currState.getCompName().equals(stateToMatch.componentName)) { + component, command, currState.getComponentName(), currState.getState()); + if (currState.getComponentName().equals(stateToMatch.componentName)) { if (currState.getState() != stateToMatch.state) { if (stateToMatch.state == State.STARTED) { log.info("Cannot schedule {} {} as dependency {} is {}", - component, command, currState.getCompName(), currState.getState()); + component, command, currState.getComponentName(), currState.getState()); canExecute = false; } else { //state is INSTALLED if (currState.getState() != State.STARTING && currState.getState() != State.STARTED) { log.info("Cannot schedule {} {} as dependency {} is {}", - component, command, currState.getCompName(), currState.getState()); + component, command, currState.getComponentName(), currState.getState()); canExecute = false; } } http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/main/java/org/apache/slider/providers/agent/ComponentInstanceState.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/providers/agent/ComponentInstanceState.java b/slider-core/src/main/java/org/apache/slider/providers/agent/ComponentInstanceState.java index 60a6f82..f7f8bf4 100644 --- a/slider-core/src/main/java/org/apache/slider/providers/agent/ComponentInstanceState.java +++ b/slider-core/src/main/java/org/apache/slider/providers/agent/ComponentInstanceState.java @@ -19,6 +19,8 @@ package org.apache.slider.providers.agent; import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import org.apache.hadoop.yarn.api.records.ContainerId; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -30,8 +32,9 @@ public class ComponentInstanceState { private static String INVALID_TRANSITION_ERROR = "Result {0} for command {1} is not expected for component {2} in state {3}."; - private final String compName; - private final String containerId; + private final String componentName; + private final ContainerId containerId; + private final String containerIdAsString; private final String applicationId; private State state = State.INIT; private State targetState = State.STARTED; @@ -40,18 +43,19 @@ public class ComponentInstanceState { private long lastHeartbeat = 0; private ContainerState containerState; - public ComponentInstanceState(String compName, - String containerId, - String applicationId) { - this.compName = compName; + public ComponentInstanceState(String componentName, + ContainerId containerId, + String applicationId) { + this.componentName = componentName; this.containerId = containerId; + this.containerIdAsString = containerId.toString(); this.applicationId = applicationId; this.containerState = ContainerState.INIT; this.lastHeartbeat = System.currentTimeMillis(); } - public String getCompName() { - return compName; + public String getComponentName() { + return componentName; } public Boolean getConfigReported() { @@ -74,20 +78,32 @@ public class ComponentInstanceState { return lastHeartbeat; } - public void setLastHeartbeat(long lastHeartbeat) { - this.lastHeartbeat = lastHeartbeat; - if(this.containerState == ContainerState.UNHEALTHY || - this.containerState == ContainerState.INIT) { - this.containerState = ContainerState.HEALTHY; + /** + * Update the heartbeat, and change container state + * to mark as healthy if appropriate + * @param heartbeatTime last time the heartbeat was seen + * @return the current container state + */ + public ContainerState heartbeat(long heartbeatTime) { + this.lastHeartbeat = heartbeatTime; + if(containerState == ContainerState.UNHEALTHY || + containerState == ContainerState.INIT) { + containerState = ContainerState.HEALTHY; } + return containerState; + } + + + public ContainerId getContainerId() { + return containerId; } public void commandIssued(Command command) { Command expected = getNextCommand(); if (expected != command) { - throw new IllegalArgumentException("Command " + command + " is not allowed is state " + state); + throw new IllegalArgumentException("Command " + command + " is not allowed in state " + state); } - this.state = this.state.getNextState(command); + state = state.getNextState(command); } public void applyCommandResult(CommandResult result, Command command) { @@ -101,12 +117,12 @@ public class ComponentInstanceState { } else if (result == CommandResult.COMPLETED) { failuresSeen = 0; } - this.state = this.state.getNextState(result); + state = state.getNextState(result); } catch (IllegalArgumentException e) { String message = String.format(INVALID_TRANSITION_ERROR, result.toString(), command.toString(), - compName, + componentName, state.toString()); log.warn(message); throw new IllegalStateException(message); @@ -114,8 +130,8 @@ public class ComponentInstanceState { } public boolean hasPendingCommand() { - if (this.state.canIssueCommands() && - this.state != this.targetState && + if (state.canIssueCommands() && + state != targetState && failuresSeen < MAX_FAILURE_TOLERATED) { return true; } @@ -144,8 +160,8 @@ public class ComponentInstanceState { public int hashCode() { int hashCode = 1; - hashCode = hashCode ^ (compName != null ? compName.hashCode() : 0); - hashCode = hashCode ^ (containerId != null ? containerId.hashCode() : 0); + hashCode = hashCode ^ (componentName != null ? componentName.hashCode() : 0); + hashCode = hashCode ^ (containerIdAsString != null ? containerIdAsString.hashCode() : 0); hashCode = hashCode ^ (applicationId != null ? applicationId.hashCode() : 0); return hashCode; } @@ -158,13 +174,13 @@ public class ComponentInstanceState { ComponentInstanceState that = (ComponentInstanceState) o; - if (this.compName != null ? - !this.compName.equals(that.compName) : this.compName != null) { + if (this.componentName != null ? + !this.componentName.equals(that.componentName) : this.componentName != null) { return false; } - if (this.containerId != null ? - !this.containerId.equals(that.containerId) : this.containerId != null) { + if (this.containerIdAsString != null ? + !this.containerIdAsString.equals(that.containerIdAsString) : this.containerIdAsString != null) { return false; } @@ -175,4 +191,18 @@ public class ComponentInstanceState { return true; } + + @Override + public String toString() { + final StringBuilder sb = + new StringBuilder("ComponentInstanceState{"); + sb.append("containerIdAsString='").append(containerIdAsString).append('\''); + sb.append(", state=").append(state); + sb.append(", failuresSeen=").append(failuresSeen); + sb.append(", lastHeartbeat=").append(lastHeartbeat); + sb.append(", containerState=").append(containerState); + sb.append(", componentName='").append(componentName).append('\''); + sb.append('}'); + return sb.toString(); + } } http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/main/java/org/apache/slider/providers/agent/HeartbeatMonitor.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/providers/agent/HeartbeatMonitor.java b/slider-core/src/main/java/org/apache/slider/providers/agent/HeartbeatMonitor.java index 3aeff66..cb22d2e 100644 --- a/slider-core/src/main/java/org/apache/slider/providers/agent/HeartbeatMonitor.java +++ b/slider-core/src/main/java/org/apache/slider/providers/agent/HeartbeatMonitor.java @@ -17,6 +17,9 @@ */ package org.apache.slider.providers.agent; +import com.google.common.annotations.VisibleForTesting; +import org.apache.hadoop.yarn.api.records.ContainerId; +import org.apache.slider.core.exceptions.SliderException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -53,10 +56,7 @@ public class HeartbeatMonitor implements Runnable { } public boolean isAlive() { - if (monitorThread != null) { - return monitorThread.isAlive(); - } - return false; + return monitorThread != null && monitorThread.isAlive(); } @Override @@ -66,7 +66,7 @@ public class HeartbeatMonitor implements Runnable { log.debug("Putting monitor to sleep for " + threadWakeupInterval + " " + "milliseconds"); Thread.sleep(threadWakeupInterval); - doWork(); + doWork(System.currentTimeMillis()); } catch (InterruptedException ex) { log.warn("Scheduler thread is interrupted going to stop", ex); shouldRun = false; @@ -83,32 +83,52 @@ public class HeartbeatMonitor implements Runnable { * received in last check interval they are marked as UNHEALTHY. INIT is when the agent is started but it did not * communicate at all. HEALTHY being the AM has received heartbeats. After an interval as UNHEALTHY the container is * declared unavailable + * @param now current time in milliseconds ... tests can set this explicitly */ - private void doWork() { + @VisibleForTesting + public void doWork(long now) { Map<String, ComponentInstanceState> componentStatuses = provider.getComponentStatuses(); if (componentStatuses != null) { for (String containerLabel : componentStatuses.keySet()) { ComponentInstanceState componentInstanceState = componentStatuses.get(containerLabel); - long timeSinceLastHeartbeat = System.currentTimeMillis() - componentInstanceState.getLastHeartbeat(); + long timeSinceLastHeartbeat = now - componentInstanceState.getLastHeartbeat(); if (timeSinceLastHeartbeat > threadWakeupInterval) { - if (componentInstanceState.getContainerState() == ContainerState.HEALTHY || - componentInstanceState.getContainerState() == ContainerState.INIT) { - componentInstanceState.setContainerState(ContainerState.UNHEALTHY); - log.warn("Component {} marked UNHEALTHY. Last heartbeat received at {} approx. {} ms. back.", - containerLabel, componentInstanceState.getLastHeartbeat(), - timeSinceLastHeartbeat); - continue; - } - if (componentInstanceState.getContainerState() == ContainerState.UNHEALTHY - && timeSinceLastHeartbeat > threadWakeupInterval * 2) { - componentInstanceState.setContainerState(ContainerState.HEARTBEAT_LOST); - log.warn("Component {} marked HEARTBEAT_LOST. Last heartbeat received at {} approx. {} ms. back.", - containerLabel, componentInstanceState.getLastHeartbeat(), - timeSinceLastHeartbeat); - this.provider.releaseContainer(containerLabel); - continue; + switch (componentInstanceState.getContainerState()) { + case INIT: + case HEALTHY: + componentInstanceState.setContainerState(ContainerState.UNHEALTHY); + log.warn( + "Component {} marked UNHEALTHY. Last heartbeat received at {} approx. {} ms. back.", + componentInstanceState, + componentInstanceState.getLastHeartbeat(), + timeSinceLastHeartbeat); + break; + case UNHEALTHY: + if (timeSinceLastHeartbeat > threadWakeupInterval * 2) { + componentInstanceState.setContainerState( + ContainerState.HEARTBEAT_LOST); + log.warn( + "Component {} marked HEARTBEAT_LOST. Last heartbeat received at {} approx. {} ms. back.", + componentInstanceState, componentInstanceState.getLastHeartbeat(), + timeSinceLastHeartbeat); + ContainerId containerId = + componentInstanceState.getContainerId(); + try { + provider.lostContainer(containerLabel, containerId); + } catch (SliderException e) { + log.info( + "Error while requesting container release for {}. Message: {}", + containerId, e, e); + } + } + break; + case HEARTBEAT_LOST: + // unexpected case + log.warn("Heartbeat from lost component: {}", componentInstanceState); + break; } + } } } http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/main/java/org/apache/slider/server/appmaster/AMViewForProviders.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/AMViewForProviders.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/AMViewForProviders.java index 287035f..f426551 100644 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/AMViewForProviders.java +++ b/slider-core/src/main/java/org/apache/slider/server/appmaster/AMViewForProviders.java @@ -18,10 +18,29 @@ package org.apache.slider.server.appmaster; +import org.apache.hadoop.yarn.api.records.ContainerId; import org.apache.slider.core.exceptions.SliderException; /** Operations available to a provider from AppMaster */ public interface AMViewForProviders { + + /** + * Outcomes from container loss + */ + enum ContainerLossReportOutcomes { + /** + * The container doesn't exist...either it wasn't in use or it + * has been released + */ + CONTAINER_NOT_IN_USE, + + /** + * The container is known about and a review has been initated + */ + CONTAINER_LOST_REVIEW_INITIATED, + } + /** Provider can ask AppMaster to release a specific container */ - void refreshContainer(String containerId, boolean newHostIfPossible) throws SliderException; + ContainerLossReportOutcomes providerLostContainer(ContainerId containerId) throws SliderException; + } http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java index 96da98d..b3650ca 100644 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java +++ b/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java @@ -1431,22 +1431,27 @@ public class SliderAppMaster extends AbstractSliderLaunchedService /* =================================================================== */ /** - * Refreshes the container by releasing it and having it reallocated - * - * @param containerId id of the container to release - * @param newHostIfPossible allocate the replacement container on a new host + * report container loss. If this isn't already known about, react * + * @param containerId id of the container which has failed * @throws SliderException */ - public void refreshContainer(String containerId, boolean newHostIfPossible) + public synchronized ContainerLossReportOutcomes providerLostContainer( + ContainerId containerId) throws SliderException { - log.info( - "Refreshing container {} per provider request.", + log.info("containerLostContactWithProvider: container {} lost", containerId); - rmOperationHandler.execute(appState.releaseContainer(containerId)); - - // ask for more containers if needed - reviewRequestAndReleaseNodes(); + RoleInstance activeContainer = appState.getActiveContainer(containerId); + if (activeContainer != null) { + rmOperationHandler.execute(appState.releaseContainer(containerId)); + // ask for more containers if needed + log.info("Container released; triggering review"); + reviewRequestAndReleaseNodes(); + return ContainerLossReportOutcomes.CONTAINER_LOST_REVIEW_INITIATED; + } else { + log.info("Container not in active set - ignoring"); + return ContainerLossReportOutcomes.CONTAINER_NOT_IN_USE; + } } /* =================================================================== */ http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/main/java/org/apache/slider/server/appmaster/state/AppState.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/AppState.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/state/AppState.java index db74ea8..08d43db 100644 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/AppState.java +++ b/slider-core/src/main/java/org/apache/slider/server/appmaster/state/AppState.java @@ -1217,7 +1217,7 @@ public class AppState { //build the failure message if (failedContainer != null) { String completedLogsUrl = getLogsURLForContainer(failedContainer); - message = String.format("Failure %s on host %s: %s" + + message = String.format("Failure %s on host %s: %s", roleInstance.getContainerId(), failedContainer.getNodeId().getHost(), completedLogsUrl); @@ -1510,12 +1510,12 @@ public class AppState { * @return * @throws SliderInternalStateException */ - public List<AbstractRMOperation> releaseContainer(String containerId) + public List<AbstractRMOperation> releaseContainer(ContainerId containerId) throws SliderInternalStateException { List<AbstractRMOperation> operations = new ArrayList<>(); List<RoleInstance> activeRoleInstances = cloneActiveContainerList(); for (RoleInstance role : activeRoleInstances) { - if (role.container.getId().toString().equals(containerId)) { + if (role.container.getId().equals(containerId)) { containerReleaseSubmitted(role.container); operations.add(new ContainerReleaseOperation(role.getId())); } http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/test/groovy/org/apache/slider/server/appmaster/model/mock/MockContainer.groovy ---------------------------------------------------------------------- diff --git a/slider-core/src/test/groovy/org/apache/slider/server/appmaster/model/mock/MockContainer.groovy b/slider-core/src/test/groovy/org/apache/slider/server/appmaster/model/mock/MockContainer.groovy index 25bee36..3eba7c4 100644 --- a/slider-core/src/test/groovy/org/apache/slider/server/appmaster/model/mock/MockContainer.groovy +++ b/slider-core/src/test/groovy/org/apache/slider/server/appmaster/model/mock/MockContainer.groovy @@ -20,7 +20,7 @@ package org.apache.slider.server.appmaster.model.mock import org.apache.hadoop.yarn.api.records.* -class MockContainer extends Container{ +class MockContainer extends Container { ContainerId id; NodeId nodeId http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java ---------------------------------------------------------------------- diff --git a/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java b/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java index 922de68..d16603f 100644 --- a/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java +++ b/slider-core/src/test/java/org/apache/slider/providers/agent/TestAgentProviderService.java @@ -21,7 +21,6 @@ package org.apache.slider.providers.agent; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.FilterFileSystem; import org.apache.hadoop.fs.Path; -import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; import org.apache.hadoop.yarn.api.records.Container; import org.apache.hadoop.yarn.api.records.ContainerId; import org.apache.hadoop.yarn.api.records.ContainerLaunchContext; @@ -85,6 +84,7 @@ import static org.easymock.EasyMock.anyObject; import static org.easymock.EasyMock.createNiceMock; import static org.easymock.EasyMock.expect; import static org.easymock.EasyMock.replay; +import static org.junit.Assert.assertEquals; import static org.mockito.Matchers.any; import static org.mockito.Matchers.anyCollection; import static org.mockito.Matchers.anyString; @@ -315,15 +315,15 @@ public class TestAgentProviderService { ClusterDescription cd = new ClusterDescription(); cd.status = new HashMap<String, Object>(); Map<String, Map<String, ClusterNode>> roleMap = new HashMap<>(); - ClusterNode cn1 = new ClusterNode(new MyContainerId(1)); + ClusterNode cn1 = new ClusterNode(new MockContainerId(1)); cn1.host = "FIRST_HOST"; Map<String, ClusterNode> map1 = new HashMap<>(); map1.put("FIRST_CONTAINER", cn1); - ClusterNode cn2 = new ClusterNode(new MyContainerId(2)); + ClusterNode cn2 = new ClusterNode(new MockContainerId(2)); cn2.host = "SECOND_HOST"; Map<String, ClusterNode> map2 = new HashMap<>(); map2.put("SECOND_CONTAINER", cn2); - ClusterNode cn3 = new ClusterNode(new MyContainerId(3)); + ClusterNode cn3 = new ClusterNode(new MockContainerId(3)); cn3.host = "THIRD_HOST"; map2.put("THIRD_CONTAINER", cn3); @@ -416,12 +416,13 @@ public class TestAgentProviderService { Map<String, Map<String, ClusterNode>> roleClusterNodeMap = new HashMap<>(); Map<String, ClusterNode> container = new HashMap<>(); - ClusterNode cn1 = new ClusterNode(new MyContainerId(1)); + ClusterNode cn1 = new ClusterNode(new MockContainerId(1)); cn1.host = "HOST1"; container.put("cid1", cn1); roleClusterNodeMap.put("HBASE_MASTER", container); - ComponentInstanceState componentStatus = new ComponentInstanceState("HBASE_MASTER", "aid", "cid"); + ComponentInstanceState componentStatus = new ComponentInstanceState("HBASE_MASTER", + new MockContainerId(1), "cid"); AgentProviderService mockAps = Mockito.spy(aps); doNothing().when(mockAps).publishApplicationInstanceData(anyString(), anyString(), anyCollection()); doReturn(metainfo).when(mockAps).getMetainfo(); @@ -572,7 +573,7 @@ public class TestAgentProviderService { } @Test - public void testOrchastratedAppStart() throws IOException { + public void testOrchestratedAppStart() throws IOException { // App has two components HBASE_MASTER and HBASE_REGIONSERVER // Start of HBASE_RS depends on the start of HBASE_MASTER InputStream metainfo_1 = new ByteArrayInputStream(metainfo_1_str.getBytes()); @@ -811,8 +812,9 @@ public class TestAgentProviderService { AgentProviderService mockAps = Mockito.spy(aps); doNothing().when(mockAps).publishApplicationInstanceData(anyString(), anyString(), anyCollection()); - ContainerId cid = new MyContainerId(1); + ContainerId cid = new MockContainerId(1); String id = cid.toString(); + ContainerId cid2 = new MockContainerId(2); mockAps.getAllocatedPorts().put("a", "100"); mockAps.getAllocatedPorts(id).put("b", "101"); mockAps.getAllocatedPorts("cid2").put("c", "102"); @@ -820,8 +822,8 @@ public class TestAgentProviderService { mockAps.getComponentInstanceData().put("cid2", new HashMap<String, String>()); mockAps.getComponentInstanceData().put(id, new HashMap<String, String>()); - mockAps.getComponentStatuses().put("cid2_HM", new ComponentInstanceState("HM", "cid2", "aid")); - mockAps.getComponentStatuses().put(id + "_HM", new ComponentInstanceState("HM", id, "aid")); + mockAps.getComponentStatuses().put("cid2_HM", new ComponentInstanceState("HM", cid2, "aid")); + mockAps.getComponentStatuses().put(id + "_HM", new ComponentInstanceState("HM", cid, "aid")); Assert.assertNotNull(mockAps.getComponentInstanceData().get(id)); Assert.assertNotNull(mockAps.getComponentInstanceData().get("cid2")); @@ -834,7 +836,7 @@ public class TestAgentProviderService { Assert.assertEquals(mockAps.getAllocatedPorts("cid2").size(), 1); // Make the call - mockAps.notifyContainerCompleted(new MyContainerId(1)); + mockAps.notifyContainerCompleted(new MockContainerId(1)); Assert.assertEquals(mockAps.getAllocatedPorts().size(), 1); Assert.assertEquals(mockAps.getAllocatedPorts(id).size(), 0); @@ -875,7 +877,7 @@ public class TestAgentProviderService { Map<String, Map<String, ClusterNode>> roleClusterNodeMap = new HashMap<>(); Map<String, ClusterNode> container = new HashMap<>(); - ClusterNode cn1 = new ClusterNode(new MyContainerId(1)); + ClusterNode cn1 = new ClusterNode(new MockContainerId(1)); cn1.host = "HOST1"; container.put("cid1", cn1); roleClusterNodeMap.put("HBASE_MASTER", container); @@ -922,7 +924,7 @@ public class TestAgentProviderService { Map<String, Map<String, ClusterNode>> roleClusterNodeMap = new HashMap<>(); Map<String, ClusterNode> container = new HashMap<>(); - ClusterNode cn1 = new ClusterNode(new MyContainerId(1)); + ClusterNode cn1 = new ClusterNode(new MockContainerId(1)); cn1.host = "HOST1"; container.put("cid1", cn1); roleClusterNodeMap.put("HBASE_MASTER", container); @@ -941,124 +943,11 @@ public class TestAgentProviderService { Assert.assertTrue(hbr.getExecutionCommands().get(0).getConfigurations().containsKey("hbase-site")); Map<String, String> hbaseSiteConf = hbr.getExecutionCommands().get(0).getConfigurations().get("hbase-site"); Assert.assertTrue(hbaseSiteConf.containsKey("a.port")); - Assert.assertTrue(hbaseSiteConf.get("a.port").equals("10023")); - Assert.assertTrue(hbaseSiteConf.get("b.port").equals("10024")); - Assert.assertTrue(hbaseSiteConf.get("random.port").equals("10025")); - Assert.assertTrue(hbaseSiteConf.get("random2.port").equals("${HBASE_MASTER.ALLOCATED_PORT}")); + Assert.assertEquals("10023", hbaseSiteConf.get("a.port")); + Assert.assertEquals("10024", hbaseSiteConf.get("b.port")); + Assert.assertEquals("10025", hbaseSiteConf.get("random.port")); + assertEquals("${HBASE_MASTER.ALLOCATED_PORT}", + hbaseSiteConf.get("random2.port")); } - private static class MyContainer extends Container { - - ContainerId cid = null; - - @Override - public ContainerId getId() { - return this.cid; - } - - @Override - public void setId(ContainerId containerId) { - this.cid = containerId; - } - - @Override - public NodeId getNodeId() { - return null; //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public void setNodeId(NodeId nodeId) { - //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public String getNodeHttpAddress() { - return null; //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public void setNodeHttpAddress(String s) { - //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public Resource getResource() { - return null; //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public void setResource(Resource resource) { - //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public Priority getPriority() { - return null; //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public void setPriority(Priority priority) { - //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public Token getContainerToken() { - return null; //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public void setContainerToken(Token token) { - //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public int compareTo(Container o) { - return 0; //To change body of implemented methods use File | Settings | File Templates. - } - } - - private static class MyContainerId extends ContainerId { - int id; - - private MyContainerId(int id) { - this.id = id; - } - - @Override - public ApplicationAttemptId getApplicationAttemptId() { - return null; //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - protected void setApplicationAttemptId(ApplicationAttemptId applicationAttemptId) { - //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public int getId() { - return id; //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - protected void setId(int i) { - //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - protected void build() { - //To change body of implemented methods use File | Settings | File Templates. - } - - @Override - public int hashCode() { - return this.id; - } - - @Override - public String toString() { - return "MyContainerId{" + - "id=" + id + - '}'; - } - } } http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/test/java/org/apache/slider/providers/agent/TestComponentCommandOrder.java ---------------------------------------------------------------------- diff --git a/slider-core/src/test/java/org/apache/slider/providers/agent/TestComponentCommandOrder.java b/slider-core/src/test/java/org/apache/slider/providers/agent/TestComponentCommandOrder.java index 3ef1839..c123fbb 100644 --- a/slider-core/src/test/java/org/apache/slider/providers/agent/TestComponentCommandOrder.java +++ b/slider-core/src/test/java/org/apache/slider/providers/agent/TestComponentCommandOrder.java @@ -19,6 +19,7 @@ package org.apache.slider.providers.agent; import org.apache.slider.providers.agent.application.metadata.CommandOrder; +import org.apache.slider.server.appmaster.model.mock.MockContainerId; import org.junit.Assert; import org.junit.Test; import org.slf4j.Logger; @@ -29,6 +30,7 @@ import java.util.Arrays; public class TestComponentCommandOrder { protected static final Logger log = LoggerFactory.getLogger(TestComponentCommandOrder.class); + private final MockContainerId containerId = new MockContainerId(1); @Test public void testComponentCommandOrder() throws Exception { @@ -43,11 +45,12 @@ public class TestComponentCommandOrder { co3.setRequires("C-STARTED,D-STARTED,E-INSTALLED"); ComponentCommandOrder cco = new ComponentCommandOrder(Arrays.asList(co1, co2, co3)); - ComponentInstanceState cisB = new ComponentInstanceState("B", "cid", "aid"); - ComponentInstanceState cisC = new ComponentInstanceState("C", "cid", "aid"); - ComponentInstanceState cisD = new ComponentInstanceState("D", "cid", "aid"); - ComponentInstanceState cisE = new ComponentInstanceState("E", "cid", "aid"); - ComponentInstanceState cisE2 = new ComponentInstanceState("E", "cid", "aid"); + ComponentInstanceState cisB = new ComponentInstanceState("B", + containerId, "aid"); + ComponentInstanceState cisC = new ComponentInstanceState("C", containerId, "aid"); + ComponentInstanceState cisD = new ComponentInstanceState("D", containerId, "aid"); + ComponentInstanceState cisE = new ComponentInstanceState("E", containerId, "aid"); + ComponentInstanceState cisE2 = new ComponentInstanceState("E", containerId, "aid"); cisB.setState(State.STARTED); cisC.setState(State.INSTALLED); Assert.assertTrue(cco.canExecute("A", Command.START, Arrays.asList(cisB))); @@ -92,8 +95,8 @@ public class TestComponentCommandOrder { co.setCommand(" A-START"); co.setRequires("B-STARTED , C-STARTED"); - ComponentInstanceState cisB = new ComponentInstanceState("B", "cid", "aid"); - ComponentInstanceState cisC = new ComponentInstanceState("C", "cid", "aid"); + ComponentInstanceState cisB = new ComponentInstanceState("B", containerId, "aid"); + ComponentInstanceState cisC = new ComponentInstanceState("C", containerId, "aid"); cisB.setState(State.STARTED); cisC.setState(State.STARTED); http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/test/java/org/apache/slider/providers/agent/TestComponentInstanceState.java ---------------------------------------------------------------------- diff --git a/slider-core/src/test/java/org/apache/slider/providers/agent/TestComponentInstanceState.java b/slider-core/src/test/java/org/apache/slider/providers/agent/TestComponentInstanceState.java index be9f178..e1f3d02 100644 --- a/slider-core/src/test/java/org/apache/slider/providers/agent/TestComponentInstanceState.java +++ b/slider-core/src/test/java/org/apache/slider/providers/agent/TestComponentInstanceState.java @@ -19,6 +19,7 @@ package org.apache.slider.providers.agent; import junit.framework.TestCase; +import org.apache.slider.server.appmaster.model.mock.MockContainerId; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -29,6 +30,7 @@ public class TestComponentInstanceState { private State[] states = new State[]{ State.INIT, State.INSTALLING, State.INSTALLED, State.STARTING, State.STARTED, State.INSTALL_FAILED}; + private final MockContainerId containerId = new MockContainerId(1); @Test public void testValidateSupportedCommands() { @@ -67,97 +69,107 @@ public class TestComponentInstanceState { @Test public void testGetNextStateBasedOnCommand() { - for (int index = 0; index < states.length; index++) { - TestCase.assertEquals(states[index], states[index].getNextState(Command.NOP)); + for (State state : states) { + TestCase.assertEquals(state, state.getNextState(Command.NOP)); } TestCase.assertEquals(State.INSTALLING, State.INIT.getNextState(Command.INSTALL)); TestCase.assertEquals(State.INSTALLING, State.INSTALL_FAILED.getNextState(Command.INSTALL)); - expectExceptionOnGetNextForCommand(IllegalArgumentException.class, State.INSTALLED, Command.INSTALL); - expectExceptionOnGetNextForCommand(IllegalArgumentException.class, State.STARTING, Command.INSTALL); - expectExceptionOnGetNextForCommand(IllegalArgumentException.class, State.STARTED, Command.INSTALL); + expectIllegalArgumentException(State.INSTALLED, Command.INSTALL); + expectIllegalArgumentException(State.STARTING, Command.INSTALL); + expectIllegalArgumentException(State.STARTED, Command.INSTALL); TestCase.assertEquals(State.STARTING, State.INSTALLED.getNextState(Command.START)); - expectExceptionOnGetNextForCommand(IllegalArgumentException.class, State.INIT, Command.START); - expectExceptionOnGetNextForCommand(IllegalArgumentException.class, State.INSTALL_FAILED, Command.START); - expectExceptionOnGetNextForCommand(IllegalArgumentException.class, State.STARTING, Command.START); - expectExceptionOnGetNextForCommand(IllegalArgumentException.class, State.INSTALLING, Command.START); - expectExceptionOnGetNextForCommand(IllegalArgumentException.class, State.STARTED, Command.START); + expectIllegalArgumentException(State.INIT, Command.START); + expectIllegalArgumentException(State.INSTALL_FAILED, Command.START); + expectIllegalArgumentException(State.STARTING, Command.START); + expectIllegalArgumentException(State.INSTALLING, Command.START); + expectIllegalArgumentException(State.STARTED, Command.START); + } + + protected void expectIllegalArgumentException(State state, Command command) { + expectExceptionOnGetNextForCommand(IllegalArgumentException.class, + state, command); } @Test public void validateStateTransitionNormal() { - ComponentInstanceState componentInstanceState = new ComponentInstanceState("HBASE_MASTER", "CID_001", "AID_001"); - TestCase.assertEquals(State.INIT, componentInstanceState.getState()); + ComponentInstanceState componentInstanceState = new ComponentInstanceState("HBASE_MASTER", containerId, "AID_001"); + assertInState(State.INIT, componentInstanceState); TestCase.assertEquals(true, componentInstanceState.hasPendingCommand()); TestCase.assertEquals(Command.INSTALL, componentInstanceState.getNextCommand()); - TestCase.assertEquals(State.INIT, componentInstanceState.getState()); + assertInState(State.INIT, componentInstanceState); componentInstanceState.commandIssued(Command.INSTALL); - TestCase.assertEquals(State.INSTALLING, componentInstanceState.getState()); + assertInState(State.INSTALLING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.IN_PROGRESS, Command.INSTALL); - TestCase.assertEquals(State.INSTALLING, componentInstanceState.getState()); + assertInState(State.INSTALLING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.COMPLETED, Command.INSTALL); - TestCase.assertEquals(State.INSTALLED, componentInstanceState.getState()); + assertInState(State.INSTALLED, componentInstanceState); TestCase.assertEquals(Command.START, componentInstanceState.getNextCommand()); componentInstanceState.commandIssued(Command.START); - TestCase.assertEquals(State.STARTING, componentInstanceState.getState()); + assertInState(State.STARTING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.IN_PROGRESS, Command.START); - TestCase.assertEquals(State.STARTING, componentInstanceState.getState()); + assertInState(State.STARTING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.COMPLETED, Command.START); - TestCase.assertEquals(State.STARTED, componentInstanceState.getState()); + assertInState(State.STARTED, componentInstanceState); + } + + protected void assertInState(State state, + ComponentInstanceState componentInstanceState) { + TestCase.assertEquals(state, componentInstanceState.getState()); } @Test public void validateStateTransitionScenario2() { - ComponentInstanceState componentInstanceState = new ComponentInstanceState("HBASE_MASTER", "CID_001", "AID_001"); - TestCase.assertEquals(State.INIT, componentInstanceState.getState()); + ComponentInstanceState componentInstanceState = new ComponentInstanceState("HBASE_MASTER", containerId, "AID_001"); + assertInState(State.INIT, componentInstanceState); TestCase.assertEquals(true, componentInstanceState.hasPendingCommand()); TestCase.assertEquals(Command.INSTALL, componentInstanceState.getNextCommand()); - TestCase.assertEquals(State.INIT, componentInstanceState.getState()); + assertInState(State.INIT, componentInstanceState); componentInstanceState.commandIssued(Command.INSTALL); - TestCase.assertEquals(State.INSTALLING, componentInstanceState.getState()); + assertInState(State.INSTALLING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.FAILED, Command.INSTALL); - TestCase.assertEquals(State.INSTALL_FAILED, componentInstanceState.getState()); + assertInState(State.INSTALL_FAILED, componentInstanceState); componentInstanceState.commandIssued(Command.INSTALL); - TestCase.assertEquals(State.INSTALLING, componentInstanceState.getState()); + assertInState(State.INSTALLING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.COMPLETED, Command.INSTALL); - TestCase.assertEquals(State.INSTALLED, componentInstanceState.getState()); + assertInState(State.INSTALLED, componentInstanceState); TestCase.assertEquals(Command.START, componentInstanceState.getNextCommand()); componentInstanceState.commandIssued(Command.START); - TestCase.assertEquals(State.STARTING, componentInstanceState.getState()); + assertInState(State.STARTING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.FAILED, Command.START); - TestCase.assertEquals(State.INSTALLED, componentInstanceState.getState()); + assertInState(State.INSTALLED, componentInstanceState); componentInstanceState.commandIssued(Command.START); componentInstanceState.applyCommandResult(CommandResult.COMPLETED, Command.START); - TestCase.assertEquals(State.STARTED, componentInstanceState.getState()); + assertInState(State.STARTED, componentInstanceState); } @Test public void tolerateMaxFailures() { - ComponentInstanceState componentInstanceState = new ComponentInstanceState("HBASE_MASTER", "CID_001", "AID_001"); - TestCase.assertEquals(State.INIT, componentInstanceState.getState()); + ComponentInstanceState componentInstanceState = new ComponentInstanceState("HBASE_MASTER", containerId, "AID_001"); + assertInState(State.INIT, componentInstanceState); TestCase.assertEquals(true, componentInstanceState.hasPendingCommand()); TestCase.assertEquals(Command.INSTALL, componentInstanceState.getNextCommand()); - TestCase.assertEquals(State.INIT, componentInstanceState.getState()); + assertInState(State.INIT, componentInstanceState); componentInstanceState.commandIssued(Command.INSTALL); - TestCase.assertEquals(State.INSTALLING, componentInstanceState.getState()); + assertInState(State.INSTALLING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.FAILED, Command.INSTALL); - TestCase.assertEquals(State.INSTALL_FAILED, componentInstanceState.getState()); + assertInState(State.INSTALL_FAILED, componentInstanceState); componentInstanceState.commandIssued(Command.INSTALL); - TestCase.assertEquals(State.INSTALLING, componentInstanceState.getState()); + assertInState(State.INSTALLING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.FAILED, Command.INSTALL); - TestCase.assertEquals(State.INSTALL_FAILED, componentInstanceState.getState()); + assertInState(State.INSTALL_FAILED, componentInstanceState); componentInstanceState.commandIssued(Command.INSTALL); - TestCase.assertEquals(State.INSTALLING, componentInstanceState.getState()); + assertInState(State.INSTALLING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.FAILED, Command.INSTALL); - TestCase.assertEquals(State.INSTALL_FAILED, componentInstanceState.getState()); + assertInState(State.INSTALL_FAILED, componentInstanceState); try { componentInstanceState.commandIssued(Command.INSTALL); @@ -168,41 +180,41 @@ public class TestComponentInstanceState { @Test public void tolerateFewFailureThenReset() { - ComponentInstanceState componentInstanceState = new ComponentInstanceState("HBASE_MASTER", "CID_001", "AID_001"); - TestCase.assertEquals(State.INIT, componentInstanceState.getState()); + ComponentInstanceState componentInstanceState = new ComponentInstanceState("HBASE_MASTER", containerId, "AID_001"); + assertInState(State.INIT, componentInstanceState); TestCase.assertEquals(true, componentInstanceState.hasPendingCommand()); TestCase.assertEquals(Command.INSTALL, componentInstanceState.getNextCommand()); - TestCase.assertEquals(State.INIT, componentInstanceState.getState()); + assertInState(State.INIT, componentInstanceState); componentInstanceState.commandIssued(Command.INSTALL); - TestCase.assertEquals(State.INSTALLING, componentInstanceState.getState()); + assertInState(State.INSTALLING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.FAILED, Command.INSTALL); - TestCase.assertEquals(State.INSTALL_FAILED, componentInstanceState.getState()); + assertInState(State.INSTALL_FAILED, componentInstanceState); componentInstanceState.commandIssued(Command.INSTALL); - TestCase.assertEquals(State.INSTALLING, componentInstanceState.getState()); + assertInState(State.INSTALLING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.FAILED, Command.INSTALL); - TestCase.assertEquals(State.INSTALL_FAILED, componentInstanceState.getState()); + assertInState(State.INSTALL_FAILED, componentInstanceState); componentInstanceState.commandIssued(Command.INSTALL); - TestCase.assertEquals(State.INSTALLING, componentInstanceState.getState()); + assertInState(State.INSTALLING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.COMPLETED, Command.INSTALL); - TestCase.assertEquals(State.INSTALLED, componentInstanceState.getState()); + assertInState(State.INSTALLED, componentInstanceState); componentInstanceState.commandIssued(Command.START); - TestCase.assertEquals(State.STARTING, componentInstanceState.getState()); + assertInState(State.STARTING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.FAILED, Command.START); - TestCase.assertEquals(State.INSTALLED, componentInstanceState.getState()); + assertInState(State.INSTALLED, componentInstanceState); componentInstanceState.commandIssued(Command.START); - TestCase.assertEquals(State.STARTING, componentInstanceState.getState()); + assertInState(State.STARTING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.FAILED, Command.START); - TestCase.assertEquals(State.INSTALLED, componentInstanceState.getState()); + assertInState(State.INSTALLED, componentInstanceState); componentInstanceState.commandIssued(Command.START); - TestCase.assertEquals(State.STARTING, componentInstanceState.getState()); + assertInState(State.STARTING, componentInstanceState); componentInstanceState.applyCommandResult(CommandResult.FAILED, Command.START); - TestCase.assertEquals(State.INSTALLED, componentInstanceState.getState()); + assertInState(State.INSTALLED, componentInstanceState); try { componentInstanceState.commandIssued(Command.START); @@ -213,7 +225,7 @@ public class TestComponentInstanceState { @Test public void testBadTransitions() { - ComponentInstanceState componentInstanceState = new ComponentInstanceState("HBASE_MASTER", "CID_001", "AID_001"); + ComponentInstanceState componentInstanceState = new ComponentInstanceState("HBASE_MASTER", containerId, "AID_001"); try { componentInstanceState.commandIssued(Command.START); @@ -268,7 +280,7 @@ public class TestComponentInstanceState { TestCase.fail("Must fail"); } catch (Exception e) { if (!expected.isInstance(e)) { - TestCase.fail("Unexpected exception " + e.getClass()); + throw e; } } } @@ -280,7 +292,7 @@ public class TestComponentInstanceState { TestCase.fail("Must fail"); } catch (Exception e) { if (!expected.isInstance(e)) { - TestCase.fail("Unexpected exception " + e.getClass()); + throw e; } } } http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/32fd3f56/slider-core/src/test/java/org/apache/slider/providers/agent/TestHeartbeatMonitor.java ---------------------------------------------------------------------- diff --git a/slider-core/src/test/java/org/apache/slider/providers/agent/TestHeartbeatMonitor.java b/slider-core/src/test/java/org/apache/slider/providers/agent/TestHeartbeatMonitor.java index c2cfafd..b16b145 100644 --- a/slider-core/src/test/java/org/apache/slider/providers/agent/TestHeartbeatMonitor.java +++ b/slider-core/src/test/java/org/apache/slider/providers/agent/TestHeartbeatMonitor.java @@ -16,6 +16,9 @@ */ package org.apache.slider.providers.agent; +import org.apache.hadoop.yarn.api.records.ContainerId; +import org.apache.slider.server.appmaster.AMViewForProviders; +import org.apache.slider.server.appmaster.model.mock.MockContainerId; import org.junit.Assert; import org.junit.Test; import org.slf4j.Logger; @@ -55,9 +58,11 @@ public class TestHeartbeatMonitor { HeartbeatMonitor hbm = new HeartbeatMonitor(provider, 500); Assert.assertFalse(hbm.isAlive()); Map<String, ComponentInstanceState> statuses = new HashMap<>(); - ComponentInstanceState state = new ComponentInstanceState("HBASE_MASTER", "Cid", "Aid"); + ContainerId container1 = new MockContainerId(1); + ComponentInstanceState state = new ComponentInstanceState("HBASE_MASTER", + container1, "Aid"); state.setState(State.STARTED); - state.setLastHeartbeat(System.currentTimeMillis()); + state.heartbeat(System.currentTimeMillis()); statuses.put("label_1", state); expect(provider.getComponentStatuses()).andReturn(statuses).anyTimes(); replay(provider); @@ -72,65 +77,100 @@ public class TestHeartbeatMonitor { @Test public void testHeartbeatMonitorWithUnhealthyAndThenLost() throws Exception { AgentProviderService provider = createNiceMock(AgentProviderService.class); - HeartbeatMonitor hbm = new HeartbeatMonitor(provider, 2 * 1000); - Assert.assertFalse(hbm.isAlive()); + long now = 100000; + int wakeupInterval = 2 * 1000; + Map<String, ComponentInstanceState> statuses = new HashMap<>(); - ComponentInstanceState masterState = new ComponentInstanceState("HBASE_MASTER", "Cid1", "Aid1"); - masterState.setState(State.STARTED); - masterState.setLastHeartbeat(System.currentTimeMillis()); - statuses.put("Aid1_Cid1_HBASE_MASTER", masterState); + ContainerId masterContainer = new MockContainerId(1); + ContainerId slaveContainer = new MockContainerId(2); + ComponentInstanceState masterState = new ComponentInstanceState("HBASE_MASTER", + masterContainer, "Aid1"); + String masterLabel = "Aid1_Cid1_HBASE_MASTER"; + statuses.put(masterLabel, masterState); + + ComponentInstanceState slaveState = new ComponentInstanceState("HBASE_REGIONSERVER", + slaveContainer, "Aid1"); + String slaveLabel = "Aid1_Cid2_HBASE_REGIONSERVER"; + statuses.put(slaveLabel, slaveState); - ComponentInstanceState slaveState = new ComponentInstanceState("HBASE_REGIONSERVER", "Cid2", "Aid1"); + masterState.setState(State.STARTED); + masterState.heartbeat(now); slaveState.setState(State.STARTED); - slaveState.setLastHeartbeat(System.currentTimeMillis()); - statuses.put("Aid1_Cid2_HBASE_REGIONSERVER", slaveState); - + slaveState.heartbeat(now); expect(provider.getComponentStatuses()).andReturn(statuses).anyTimes(); - expect(provider.releaseContainer("Aid1_Cid2_HBASE_REGIONSERVER")).andReturn(true).once(); + expect(provider.lostContainer(slaveLabel, slaveContainer)).andReturn( + AMViewForProviders.ContainerLossReportOutcomes.CONTAINER_LOST_REVIEW_INITIATED).once(); + // expect the second iteration to report no container any more + expect(provider.lostContainer(slaveLabel, slaveContainer)).andReturn( + AMViewForProviders.ContainerLossReportOutcomes.CONTAINER_NOT_IN_USE).once(); replay(provider); - hbm.start(); - Thread.sleep(1 * 1000); + + HeartbeatMonitor heartbeatMonitor = new HeartbeatMonitor(provider, + wakeupInterval); + Assert.assertFalse(heartbeatMonitor.isAlive()); + now += wakeupInterval; + masterState.setState(State.STARTED); + masterState.heartbeat(now); + + slaveState.setState(State.STARTED); // just dial back by at least 2 sec but no more than 4 - slaveState.setLastHeartbeat(System.currentTimeMillis() - (2 * 1000 + 100)); - masterState.setLastHeartbeat(System.currentTimeMillis()); + slaveState.heartbeat(now - (wakeupInterval + 100)); - Thread.sleep(1 * 1000 + 500); - masterState.setLastHeartbeat(System.currentTimeMillis()); - log.info("Slave container state {}", slaveState.getContainerState()); - Assert.assertEquals(ContainerState.HEALTHY, masterState.getContainerState()); - Assert.assertEquals(ContainerState.UNHEALTHY, slaveState.getContainerState()); + assertInState(ContainerState.HEALTHY, masterState, now); + assertInState(ContainerState.HEALTHY, slaveState, now); + + //tick #1 + heartbeatMonitor.doWork(now); - Thread.sleep(1 * 1000); - // some lost heartbeats are ignored (e.g. ~ 1 sec) - masterState.setLastHeartbeat(System.currentTimeMillis() - 1 * 1000); + assertInState(ContainerState.HEALTHY, masterState, now); + assertInState(ContainerState.UNHEALTHY, slaveState, now); - Thread.sleep(1 * 1000 + 500); + // heartbeat from the master + masterState.heartbeat(now + 1500); - log.info("Slave container state {}", slaveState.getContainerState()); - Assert.assertEquals(ContainerState.HEALTHY, masterState.getContainerState()); - Assert.assertEquals(ContainerState.HEARTBEAT_LOST, slaveState.getContainerState()); - hbm.shutdown(); + // tick #2 + now += wakeupInterval; + heartbeatMonitor.doWork(now); + + assertInState(ContainerState.HEALTHY, masterState, now); + assertInState(ContainerState.HEARTBEAT_LOST, slaveState, now); + } + + protected void assertInState(ContainerState expectedState, + ComponentInstanceState componentInstanceState, long now) { + ContainerState actualState = componentInstanceState.getContainerState(); + if (!expectedState.equals(actualState)) { + // mismatch + Assert.fail(String.format("at [%06d] Expected component state %s " + + "but found state %s in in component %s", + now, expectedState, actualState, componentInstanceState)); + } } @Test public void testHeartbeatTransitions() { - ComponentInstanceState slaveState = new ComponentInstanceState("HBASE_REGIONSERVER", "Cid2", "Aid1"); + ContainerId container2 = new MockContainerId(2); + ComponentInstanceState slaveState = new ComponentInstanceState("HBASE_REGIONSERVER", + container2, "Aid1"); slaveState.setState(State.STARTED); - Assert.assertEquals(ContainerState.INIT, slaveState.getContainerState()); - slaveState.setLastHeartbeat(System.currentTimeMillis()); - Assert.assertEquals(ContainerState.HEALTHY, slaveState.getContainerState()); + long lastHeartbeat = System.currentTimeMillis(); + assertInState(ContainerState.INIT, slaveState, 0); + slaveState.heartbeat(lastHeartbeat); + assertInState(ContainerState.HEALTHY, slaveState, lastHeartbeat); slaveState.setContainerState(ContainerState.UNHEALTHY); - Assert.assertEquals(ContainerState.UNHEALTHY, slaveState.getContainerState()); - slaveState.setLastHeartbeat(System.currentTimeMillis()); - Assert.assertEquals(ContainerState.HEALTHY, slaveState.getContainerState()); + lastHeartbeat = System.currentTimeMillis(); + assertInState(ContainerState.UNHEALTHY, slaveState, lastHeartbeat); + slaveState.heartbeat(lastHeartbeat); + assertInState(ContainerState.HEALTHY, slaveState, lastHeartbeat); slaveState.setContainerState(ContainerState.HEARTBEAT_LOST); - Assert.assertEquals(ContainerState.HEARTBEAT_LOST, slaveState.getContainerState()); - slaveState.setLastHeartbeat(System.currentTimeMillis()); - Assert.assertEquals(ContainerState.HEARTBEAT_LOST, slaveState.getContainerState()); + assertInState(ContainerState.HEARTBEAT_LOST, slaveState, lastHeartbeat); + lastHeartbeat = System.currentTimeMillis(); + slaveState.heartbeat(lastHeartbeat); + assertInState(ContainerState.HEARTBEAT_LOST, slaveState, lastHeartbeat); } }
