TestNMLeveldbStateStoreService.java
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.hadoop.yarn.server.nodemanager.recovery;
import static org.assertj.core.api.Assertions.assertThat;
import static org.fusesource.leveldbjni.JniDBFactory.bytes;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.isNull;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.service.ServiceStateException;
import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ContainerExitStatus;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.Token;
import org.apache.hadoop.yarn.api.records.URL;
import org.apache.hadoop.yarn.api.records.impl.pb.ApplicationIdPBImpl;
import org.apache.hadoop.yarn.api.records.impl.pb.LocalResourcePBImpl;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.proto.YarnProtos.LocalResourceProto;
import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.ContainerManagerApplicationProto;
import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.DeletionServiceDeleteTaskProto;
import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.LocalizedResourceProto;
import org.apache.hadoop.yarn.proto.YarnServerNodemanagerRecoveryProtos.LogDeleterProto;
import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
import org.apache.hadoop.yarn.server.api.records.MasterKey;
import org.apache.hadoop.yarn.server.nodemanager.amrmproxy.AMRMProxyTokenSecretManager;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ResourceMappings;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.resourceplugin.fpga.FpgaDevice;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.numa.NumaResourceAllocation;
import org.apache.hadoop.yarn.server.nodemanager.containermanager.resourceplugin.gpu.GpuDevice;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.LocalResourceTrackerState;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredAMRMProxyState;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredApplicationsState;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerState;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerStatus;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerTokensState;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredContainerType;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredDeletionServiceState;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredLocalizationState;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredLogDeleterState;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredNMTokensState;
import org.apache.hadoop.yarn.server.nodemanager.recovery.NMStateStoreService.RecoveredUserResources;
import org.apache.hadoop.yarn.server.records.Version;
import org.apache.hadoop.yarn.server.security.BaseContainerTokenSecretManager;
import org.apache.hadoop.yarn.server.security.BaseNMTokenSecretManager;
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
import org.iq80.leveldb.DB;
import org.iq80.leveldb.DBException;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
public class TestNMLeveldbStateStoreService {
private static final File TMP_DIR = new File(
System.getProperty("test.build.data",
System.getProperty("java.io.tmpdir")),
TestNMLeveldbStateStoreService.class.getName());
YarnConfiguration conf;
NMLeveldbStateStoreService stateStore;
@Before
public void setup() throws IOException {
FileUtil.fullyDelete(TMP_DIR);
conf = new YarnConfiguration();
conf.setBoolean(YarnConfiguration.NM_RECOVERY_ENABLED, true);
conf.set(YarnConfiguration.NM_RECOVERY_DIR, TMP_DIR.toString());
restartStateStore();
}
@After
public void cleanup() throws IOException {
if (stateStore != null) {
stateStore.close();
}
FileUtil.fullyDelete(TMP_DIR);
}
private List<RecoveredContainerState> loadContainersState(
RecoveryIterator<RecoveredContainerState> it) throws IOException {
List<RecoveredContainerState> containers =
new ArrayList<RecoveredContainerState>();
while (it.hasNext()) {
RecoveredContainerState rcs = it.next();
containers.add(rcs);
}
return containers;
}
private List<ContainerManagerApplicationProto> loadApplicationProtos(
RecoveryIterator<ContainerManagerApplicationProto> it)
throws IOException {
List<ContainerManagerApplicationProto> applicationProtos =
new ArrayList<ContainerManagerApplicationProto>();
while (it.hasNext()) {
applicationProtos.add(it.next());
}
return applicationProtos;
}
private List<DeletionServiceDeleteTaskProto> loadDeletionTaskProtos(
RecoveryIterator<DeletionServiceDeleteTaskProto> it) throws IOException {
List<DeletionServiceDeleteTaskProto> deleteTaskProtos =
new ArrayList<DeletionServiceDeleteTaskProto>();
while (it.hasNext()) {
deleteTaskProtos.add(it.next());
}
return deleteTaskProtos;
}
private Map<String, RecoveredUserResources> loadUserResources(
RecoveryIterator<Map.Entry<String, RecoveredUserResources>> it)
throws IOException {
Map<String, RecoveredUserResources> userResources =
new HashMap<String, RecoveredUserResources>();
while (it.hasNext()) {
Map.Entry<String, RecoveredUserResources> entry = it.next();
userResources.put(entry.getKey(), entry.getValue());
}
return userResources;
}
private Map<ApplicationAttemptId, MasterKey> loadNMTokens(
RecoveryIterator<Map.Entry<ApplicationAttemptId, MasterKey>> it)
throws IOException {
Map<ApplicationAttemptId, MasterKey> nmTokens =
new HashMap<ApplicationAttemptId, MasterKey>();
while (it.hasNext()) {
Map.Entry<ApplicationAttemptId, MasterKey> entry = it.next();
nmTokens.put(entry.getKey(), entry.getValue());
}
return nmTokens;
}
private Map<ContainerId, Long> loadContainerTokens(
RecoveryIterator<Map.Entry<ContainerId, Long>> it) throws IOException {
Map<ContainerId, Long> containerTokens =
new HashMap<ContainerId, Long>();
while (it.hasNext()) {
Map.Entry<ContainerId, Long> entry = it.next();
containerTokens.put(entry.getKey(), entry.getValue());
}
return containerTokens;
}
private List<LocalizedResourceProto> loadCompletedResources(
RecoveryIterator<LocalizedResourceProto> it) throws IOException {
List<LocalizedResourceProto> completedResources =
new ArrayList<LocalizedResourceProto>();
while (it != null && it.hasNext()) {
completedResources.add(it.next());
}
return completedResources;
}
private Map<LocalResourceProto, Path> loadStartedResources(
RecoveryIterator <Map.Entry<LocalResourceProto, Path>> it)
throws IOException {
Map<LocalResourceProto, Path> startedResources =
new HashMap<LocalResourceProto, Path>();
while (it != null &&it.hasNext()) {
Map.Entry<LocalResourceProto, Path> entry = it.next();
startedResources.put(entry.getKey(), entry.getValue());
}
return startedResources;
}
private void restartStateStore() throws IOException {
// need to close so leveldb releases database lock
if (stateStore != null) {
stateStore.close();
}
stateStore = new NMLeveldbStateStoreService();
stateStore.init(conf);
stateStore.start();
}
private void verifyEmptyState() throws IOException {
RecoveredLocalizationState state = stateStore.loadLocalizationState();
assertNotNull(state);
LocalResourceTrackerState pubts = state.getPublicTrackerState();
assertNotNull(pubts);
assertTrue(loadCompletedResources(pubts.getCompletedResourcesIterator())
.isEmpty());
assertTrue(loadStartedResources(pubts.getStartedResourcesIterator())
.isEmpty());
assertTrue(loadUserResources(state.getIterator()).isEmpty());
}
@Test
public void testIsNewlyCreated() throws IOException {
assertTrue(stateStore.isNewlyCreated());
restartStateStore();
assertFalse(stateStore.isNewlyCreated());
}
@Test
public void testEmptyState() throws IOException {
assertTrue(stateStore.canRecover());
verifyEmptyState();
}
@Test
public void testCheckVersion() throws IOException {
// default version
Version defaultVersion = stateStore.getCurrentVersion();
Assert.assertEquals(defaultVersion, stateStore.loadVersion());
// compatible version
Version compatibleVersion =
Version.newInstance(defaultVersion.getMajorVersion(),
defaultVersion.getMinorVersion() + 2);
stateStore.storeVersion(compatibleVersion);
Assert.assertEquals(compatibleVersion, stateStore.loadVersion());
restartStateStore();
// overwrite the compatible version
Assert.assertEquals(defaultVersion, stateStore.loadVersion());
// incompatible version
Version incompatibleVersion =
Version.newInstance(defaultVersion.getMajorVersion() + 1,
defaultVersion.getMinorVersion());
stateStore.storeVersion(incompatibleVersion);
try {
restartStateStore();
Assert.fail("Incompatible version, should expect fail here.");
} catch (ServiceStateException e) {
Assert.assertTrue("Exception message mismatch",
e.getMessage().contains("Incompatible version for NM state:"));
}
}
@Test
public void testApplicationStorage() throws IOException {
// test empty when no state
RecoveredApplicationsState state = stateStore.loadApplicationsState();
List<ContainerManagerApplicationProto> apps =
loadApplicationProtos(state.getIterator());
assertTrue(apps.isEmpty());
// store an application and verify recovered
final ApplicationId appId1 = ApplicationId.newInstance(1234, 1);
ContainerManagerApplicationProto.Builder builder =
ContainerManagerApplicationProto.newBuilder();
builder.setId(((ApplicationIdPBImpl) appId1).getProto());
builder.setUser("user1");
ContainerManagerApplicationProto appProto1 = builder.build();
stateStore.storeApplication(appId1, appProto1);
restartStateStore();
state = stateStore.loadApplicationsState();
apps = loadApplicationProtos(state.getIterator());
assertEquals(1, apps.size());
assertEquals(appProto1, apps.get(0));
// add a new app
final ApplicationId appId2 = ApplicationId.newInstance(1234, 2);
builder = ContainerManagerApplicationProto.newBuilder();
builder.setId(((ApplicationIdPBImpl) appId2).getProto());
builder.setUser("user2");
ContainerManagerApplicationProto appProto2 = builder.build();
stateStore.storeApplication(appId2, appProto2);
restartStateStore();
state = stateStore.loadApplicationsState();
apps = loadApplicationProtos(state.getIterator());
assertEquals(2, apps.size());
assertTrue(apps.contains(appProto1));
assertTrue(apps.contains(appProto2));
// test removing an application
stateStore.removeApplication(appId2);
restartStateStore();
state = stateStore.loadApplicationsState();
apps = loadApplicationProtos(state.getIterator());
assertEquals(1, apps.size());
assertEquals(appProto1, apps.get(0));
}
@Test
public void testContainerStorageWhenContainerIsRequested()
throws IOException {
final ContainerStateConstructParams containerParams =
storeContainerInStateStore();
restartStateStore();
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertEquals(1, recoveredContainers.size());
final RecoveredContainerState rcs = recoveredContainers.get(0);
assertEquals(0, rcs.getVersion());
assertEquals(containerParams.getContainerStartTime().longValue(),
rcs.getStartTime());
assertEquals(RecoveredContainerStatus.REQUESTED, rcs.getStatus());
assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode());
assertEquals(false, rcs.getKilled());
assertEquals(containerParams.getContainerRequest(), rcs.getStartRequest());
assertTrue(rcs.getDiagnostics().isEmpty());
assertEquals(containerParams.getContainerResource(), rcs.getCapability());
}
@Test
public void testContainerStorageWhenContainerIsQueued()
throws IOException {
ContainerStateConstructParams containerParams =
storeContainerInStateStore();
ContainerId containerId = containerParams.getContainerId();
StartContainerRequest containerReq = containerParams.getContainerRequest();
Resource containerResource = containerParams.getContainerResource();
ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId();
storeNewContainerRecordWithoutStartContainerRequest(appAttemptId);
stateStore.storeContainerQueued(containerId);
restartStateStore();
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertEquals(1, recoveredContainers.size());
RecoveredContainerState rcs = recoveredContainers.get(0);
assertEquals(RecoveredContainerStatus.QUEUED, rcs.getStatus());
assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode());
assertEquals(false, rcs.getKilled());
assertEquals(containerReq, rcs.getStartRequest());
assertTrue(rcs.getDiagnostics().isEmpty());
assertEquals(containerResource, rcs.getCapability());
}
@Test
public void testContainerStorageWhenContainerIsLaunched()
throws IOException {
ContainerStateConstructParams containerParams =
storeContainerInStateStore();
ContainerId containerId = containerParams.getContainerId();
StartContainerRequest containerReq = containerParams.getContainerRequest();
Resource containerResource = containerParams.getContainerResource();
ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId();
storeNewContainerRecordWithoutStartContainerRequest(appAttemptId);
stateStore.storeContainerQueued(containerId);
StringBuilder diags = launchContainerWithDiagnostics(containerId);
restartStateStore();
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertEquals(1, recoveredContainers.size());
RecoveredContainerState rcs = recoveredContainers.get(0);
assertEquals(RecoveredContainerStatus.LAUNCHED, rcs.getStatus());
assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode());
assertEquals(false, rcs.getKilled());
assertEquals(containerReq, rcs.getStartRequest());
assertEquals(diags.toString(), rcs.getDiagnostics());
assertEquals(containerResource, rcs.getCapability());
}
@Test
public void testContainerStorageWhenContainerIsPaused()
throws IOException {
ContainerStateConstructParams containerParams =
storeContainerInStateStore();
ContainerId containerId = containerParams.getContainerId();
StartContainerRequest containerReq = containerParams.getContainerRequest();
ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId();
storeNewContainerRecordWithoutStartContainerRequest(appAttemptId);
stateStore.storeContainerQueued(containerId);
stateStore.storeContainerPaused(containerId);
restartStateStore();
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertEquals(1, recoveredContainers.size());
RecoveredContainerState rcs = recoveredContainers.get(0);
assertEquals(RecoveredContainerStatus.PAUSED, rcs.getStatus());
assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode());
assertEquals(false, rcs.getKilled());
assertEquals(containerReq, rcs.getStartRequest());
// Resume the container
stateStore.removeContainerPaused(containerId);
restartStateStore();
recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertEquals(1, recoveredContainers.size());
}
@Test
public void testContainerStorageWhenContainerSizeIncreased()
throws IOException {
ContainerStateConstructParams containerParams =
storeContainerInStateStore();
ContainerId containerId = containerParams.getContainerId();
ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId();
storeNewContainerRecordWithoutStartContainerRequest(appAttemptId);
stateStore.storeContainerQueued(containerId);
launchContainerWithDiagnostics(containerId);
increaseContainerSize(containerId);
restartStateStore();
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertEquals(1, recoveredContainers.size());
RecoveredContainerState rcs = recoveredContainers.get(0);
assertEquals(0, rcs.getVersion());
assertEquals(RecoveredContainerStatus.LAUNCHED, rcs.getStatus());
assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode());
assertEquals(false, rcs.getKilled());
assertEquals(Resource.newInstance(2468, 4), rcs.getCapability());
}
@Test
public void testContainerStorageWhenContainerMarkedAsKilled()
throws IOException {
ContainerStateConstructParams containerParams =
storeContainerInStateStore();
ContainerId containerId = containerParams.getContainerId();
ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId();
storeNewContainerRecordWithoutStartContainerRequest(appAttemptId);
stateStore.storeContainerQueued(containerId);
StringBuilder diags = launchContainerWithDiagnostics(containerId);
ContainerTokenIdentifier updateTokenIdentifier =
increaseContainerSize(containerId);
markContainerAsKilled(containerId, diags);
restartStateStore();
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertEquals(1, recoveredContainers.size());
RecoveredContainerState rcs = recoveredContainers.get(0);
assertEquals(RecoveredContainerStatus.LAUNCHED, rcs.getStatus());
assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode());
assertTrue(rcs.getKilled());
ContainerTokenIdentifier tokenReadFromRequest = BuilderUtils
.newContainerTokenIdentifier(rcs.getStartRequest()
.getContainerToken());
assertEquals(updateTokenIdentifier, tokenReadFromRequest);
assertEquals(diags.toString(), rcs.getDiagnostics());
}
@Test
public void testContainerStorageWhenContainerCompleted()
throws IOException {
ContainerStateConstructParams containerParams =
storeContainerInStateStore();
ContainerId containerId = containerParams.getContainerId();
ApplicationAttemptId appAttemptId = containerParams.getAppAttemptId();
storeNewContainerRecordWithoutStartContainerRequest(appAttemptId);
stateStore.storeContainerQueued(containerId);
StringBuilder diags = launchContainerWithDiagnostics(containerId);
markContainerAsKilled(containerId, diags);
// add yet more diags, mark container completed
diags.append("some final diags");
stateStore.storeContainerDiagnostics(containerId, diags);
stateStore.storeContainerCompleted(containerId, 21);
restartStateStore();
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertEquals(1, recoveredContainers.size());
RecoveredContainerState rcs = recoveredContainers.get(0);
assertEquals(RecoveredContainerStatus.COMPLETED, rcs.getStatus());
assertEquals(21, rcs.getExitCode());
assertTrue(rcs.getKilled());
assertEquals(diags.toString(), rcs.getDiagnostics());
}
@Test
public void testContainerStorage() throws IOException {
ContainerStateConstructParams containerParams =
storeContainerInStateStore();
ContainerId containerId = containerParams.getContainerId();
// remaining retry attempts, work dir and log dir are stored
stateStore.storeContainerRemainingRetryAttempts(containerId, 6);
stateStore.storeContainerWorkDir(containerId, "/test/workdir");
stateStore.storeContainerLogDir(containerId, "/test/logdir");
restartStateStore();
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertEquals(1, recoveredContainers.size());
RecoveredContainerState rcs = recoveredContainers.get(0);
assertEquals(6, rcs.getRemainingRetryAttempts());
assertEquals("/test/workdir", rcs.getWorkDir());
assertEquals("/test/logdir", rcs.getLogDir());
validateRetryAttempts(containerId);
}
@Test
public void testContainerStorageWhenContainerRemoved()
throws IOException {
ContainerStateConstructParams containerParams =
storeContainerInStateStore();
ContainerId containerId = containerParams.getContainerId();
// remove the container and verify not recovered
stateStore.removeContainer(containerId);
restartStateStore();
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertTrue(recoveredContainers.isEmpty());
// recover again to check remove clears all containers
restartStateStore();
NMStateStoreService nmStoreSpy = spy(stateStore);
loadContainersState(nmStoreSpy.getContainerStateIterator());
verify(nmStoreSpy, times(0)).removeContainer(any(ContainerId.class));
}
private ContainerStateConstructParams storeContainerInStateStore()
throws IOException {
// test empty when no state
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertTrue(recoveredContainers.isEmpty());
// create a container request
ApplicationId appId = ApplicationId.newInstance(1234, 3);
ApplicationAttemptId appAttemptId =
ApplicationAttemptId.newInstance(appId, 4);
ContainerId containerId = ContainerId.newContainerId(appAttemptId, 5);
Resource containerResource = Resource.newInstance(1024, 2);
StartContainerRequest containerReq =
createContainerRequest(containerId, containerResource);
long anyContainerStartTime = 1573155078494L;
stateStore.storeContainer(containerId, 0, anyContainerStartTime,
containerReq);
// verify the container version key is not stored for new containers
DB db = stateStore.getDB();
assertNull("version key present for new container", db.get(bytes(
stateStore.getContainerVersionKey(containerId.toString()))));
return new ContainerStateConstructParams()
.setContainerRequest(containerReq)
.setContainerResource(containerResource)
.setContainerStartTime(anyContainerStartTime)
.setAppAttemptId(appAttemptId)
.setContainerId(containerId);
}
private static class ContainerStateConstructParams {
private StartContainerRequest containerRequest;
private Resource containerResource;
private Long containerStartTime;
private ApplicationAttemptId appAttemptId;
private ContainerId containerId;
public ApplicationAttemptId getAppAttemptId() {
return appAttemptId;
}
public ContainerStateConstructParams setAppAttemptId(ApplicationAttemptId
theAppAttemptId) {
this.appAttemptId = theAppAttemptId;
return this;
}
public ContainerId getContainerId() {
return containerId;
}
public ContainerStateConstructParams setContainerId(ContainerId
theContainerId) {
this.containerId = theContainerId;
return this;
}
public StartContainerRequest getContainerRequest() {
return containerRequest;
}
public ContainerStateConstructParams setContainerRequest(
StartContainerRequest theContainerRequest) {
this.containerRequest = theContainerRequest;
return this;
}
public Resource getContainerResource() {
return containerResource;
}
public ContainerStateConstructParams setContainerResource(
Resource theContainerResource) {
this.containerResource = theContainerResource;
return this;
}
public Long getContainerStartTime() {
return containerStartTime;
}
public ContainerStateConstructParams setContainerStartTime(
Long theContainerStartTime) {
this.containerStartTime = theContainerStartTime;
return this;
}
}
private void markContainerAsKilled(ContainerId containerId,
StringBuilder diags) throws IOException {
// mark the container killed, add some more diags
diags.append("some more diags for container");
stateStore.storeContainerDiagnostics(containerId, diags);
stateStore.storeContainerKilled(containerId);
}
private ContainerTokenIdentifier increaseContainerSize(
ContainerId containerId) throws IOException {
ContainerTokenIdentifier updateTokenIdentifier =
new ContainerTokenIdentifier(containerId, "host", "user",
Resource.newInstance(2468, 4), 9876543210L, 42, 2468,
Priority.newInstance(7), 13579);
stateStore
.storeContainerUpdateToken(containerId, updateTokenIdentifier);
return updateTokenIdentifier;
}
private StringBuilder launchContainerWithDiagnostics(ContainerId containerId)
throws IOException {
StringBuilder diags = new StringBuilder();
stateStore.storeContainerLaunched(containerId);
diags.append("some diags for container");
stateStore.storeContainerDiagnostics(containerId, diags);
return diags;
}
private void storeNewContainerRecordWithoutStartContainerRequest(
ApplicationAttemptId appAttemptId) throws IOException {
// store a new container record without StartContainerRequest
ContainerId containerId1 = ContainerId.newContainerId(appAttemptId, 6);
stateStore.storeContainerLaunched(containerId1);
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
// check whether the new container record is discarded
assertEquals(1, recoveredContainers.size());
}
private void validateRetryAttempts(ContainerId containerId)
throws IOException {
// store finishTimeForRetryAttempts
List<Long> finishTimeForRetryAttempts = Arrays.asList(1462700529039L,
1462700529050L, 1462700529120L);
stateStore.storeContainerRestartTimes(containerId,
finishTimeForRetryAttempts);
restartStateStore();
RecoveredContainerState rcs =
loadContainersState(stateStore.getContainerStateIterator()).get(0);
List<Long> recoveredRestartTimes = rcs.getRestartTimes();
assertEquals(1462700529039L, (long)recoveredRestartTimes.get(0));
assertEquals(1462700529050L, (long)recoveredRestartTimes.get(1));
assertEquals(1462700529120L, (long)recoveredRestartTimes.get(2));
}
private StartContainerRequest createContainerRequest(
ContainerId containerId, Resource res) {
return createContainerRequestInternal(containerId, res);
}
private StartContainerRequest createContainerRequestInternal(ContainerId
containerId, Resource res) {
LocalResource lrsrc = LocalResource.newInstance(
URL.newInstance("hdfs", "somehost", 12345, "/some/path/to/rsrc"),
LocalResourceType.FILE, LocalResourceVisibility.APPLICATION, 123L,
1234567890L);
Map<String, LocalResource> localResources =
new HashMap<String, LocalResource>();
localResources.put("rsrc", lrsrc);
Map<String, String> env = new HashMap<String, String>();
env.put("somevar", "someval");
List<String> containerCmds = new ArrayList<String>();
containerCmds.add("somecmd");
containerCmds.add("somearg");
Map<String, ByteBuffer> serviceData = new HashMap<String, ByteBuffer>();
serviceData.put("someservice",
ByteBuffer.wrap(new byte[] {0x1, 0x2, 0x3}));
ByteBuffer containerTokens =
ByteBuffer.wrap(new byte[] {0x7, 0x8, 0x9, 0xa});
Map<ApplicationAccessType, String> acls =
new HashMap<ApplicationAccessType, String>();
acls.put(ApplicationAccessType.VIEW_APP, "viewuser");
acls.put(ApplicationAccessType.MODIFY_APP, "moduser");
ContainerLaunchContext clc = ContainerLaunchContext.newInstance(
localResources, env, containerCmds, serviceData, containerTokens,
acls);
Resource containerRsrc = Resource.newInstance(1357, 3);
if (res != null) {
containerRsrc = res;
}
ContainerTokenIdentifier containerTokenId =
new ContainerTokenIdentifier(containerId, "host", "user",
containerRsrc, 9876543210L, 42, 2468, Priority.newInstance(7),
13579);
Token containerToken = Token.newInstance(containerTokenId.getBytes(),
ContainerTokenIdentifier.KIND.toString(), "password".getBytes(),
"tokenservice");
return StartContainerRequest.newInstance(clc, containerToken);
}
@Test
public void testLocalTrackerStateIterator() throws IOException {
String user1 = "somebody";
ApplicationId appId1 = ApplicationId.newInstance(1, 1);
ApplicationId appId2 = ApplicationId.newInstance(2, 2);
String user2 = "someone";
ApplicationId appId3 = ApplicationId.newInstance(3, 3);
// start and finish local resource for applications
Path appRsrcPath1 = new Path("hdfs://some/app/resource1");
LocalResourcePBImpl rsrcPb1 = (LocalResourcePBImpl)
LocalResource.newInstance(
URL.fromPath(appRsrcPath1),
LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION,
123L, 456L);
LocalResourceProto appRsrcProto1 = rsrcPb1.getProto();
Path appRsrcLocalPath1 = new Path("/some/local/dir/for/apprsrc1");
Path appRsrcPath2 = new Path("hdfs://some/app/resource2");
LocalResourcePBImpl rsrcPb2 = (LocalResourcePBImpl)
LocalResource.newInstance(
URL.fromPath(appRsrcPath2),
LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION,
123L, 456L);
LocalResourceProto appRsrcProto2 = rsrcPb2.getProto();
Path appRsrcLocalPath2 = new Path("/some/local/dir/for/apprsrc2");
Path appRsrcPath3 = new Path("hdfs://some/app/resource3");
LocalResourcePBImpl rsrcPb3 = (LocalResourcePBImpl)
LocalResource.newInstance(
URL.fromPath(appRsrcPath3),
LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION,
123L, 456L);
LocalResourceProto appRsrcProto3 = rsrcPb3.getProto();
Path appRsrcLocalPath3 = new Path("/some/local/dir/for/apprsrc2");
stateStore.startResourceLocalization(user1, appId1, appRsrcProto1,
appRsrcLocalPath1);
stateStore.startResourceLocalization(user1, appId2, appRsrcProto2,
appRsrcLocalPath2);
stateStore.startResourceLocalization(user2, appId3, appRsrcProto3,
appRsrcLocalPath3);
LocalizedResourceProto appLocalizedProto1 =
LocalizedResourceProto.newBuilder()
.setResource(appRsrcProto1)
.setLocalPath(appRsrcLocalPath1.toString())
.setSize(1234567L)
.build();
LocalizedResourceProto appLocalizedProto2 =
LocalizedResourceProto.newBuilder()
.setResource(appRsrcProto2)
.setLocalPath(appRsrcLocalPath2.toString())
.setSize(1234567L)
.build();
LocalizedResourceProto appLocalizedProto3 =
LocalizedResourceProto.newBuilder()
.setResource(appRsrcProto3)
.setLocalPath(appRsrcLocalPath3.toString())
.setSize(1234567L)
.build();
stateStore.finishResourceLocalization(user1, appId1, appLocalizedProto1);
stateStore.finishResourceLocalization(user1, appId2, appLocalizedProto2);
stateStore.finishResourceLocalization(user2, appId3, appLocalizedProto3);
List<LocalizedResourceProto> completedResources =
new ArrayList<LocalizedResourceProto>();
Map<LocalResourceProto, Path> startedResources =
new HashMap<LocalResourceProto, Path>();
// restart and verify two users exist and two apps completed for user1.
restartStateStore();
RecoveredLocalizationState state = stateStore.loadLocalizationState();
Map<String, RecoveredUserResources> userResources =
loadUserResources(state.getIterator());
assertEquals(2, userResources.size());
RecoveredUserResources uResource = userResources.get(user1);
assertEquals(2, uResource.getAppTrackerStates().size());
LocalResourceTrackerState app1ts =
uResource.getAppTrackerStates().get(appId1);
assertNotNull(app1ts);
completedResources = loadCompletedResources(
app1ts.getCompletedResourcesIterator());
startedResources = loadStartedResources(
app1ts.getStartedResourcesIterator());
assertTrue(startedResources.isEmpty());
assertEquals(1, completedResources.size());
assertEquals(appLocalizedProto1,
completedResources.iterator().next());
LocalResourceTrackerState app2ts =
uResource.getAppTrackerStates().get(appId2);
assertNotNull(app2ts);
completedResources = loadCompletedResources(
app2ts.getCompletedResourcesIterator());
startedResources = loadStartedResources(
app2ts.getStartedResourcesIterator());
assertTrue(startedResources.isEmpty());
assertEquals(1, completedResources.size());
assertEquals(appLocalizedProto2,
completedResources.iterator().next());
}
@Test
public void testStartResourceLocalizationForApplicationResource()
throws IOException {
String user = "somebody";
ApplicationId appId = ApplicationId.newInstance(1, 1);
// start a local resource for an application
Path appRsrcPath = new Path("hdfs://some/app/resource");
LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl)
LocalResource.newInstance(
URL.fromPath(appRsrcPath),
LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION,
123L, 456L);
LocalResourceProto appRsrcProto = rsrcPb.getProto();
Path appRsrcLocalPath = new Path("/some/local/dir/for/apprsrc");
stateStore.startResourceLocalization(user, appId, appRsrcProto,
appRsrcLocalPath);
List<LocalizedResourceProto> completedResources =
new ArrayList<LocalizedResourceProto>();
Map<LocalResourceProto, Path> startedResources =
new HashMap<LocalResourceProto, Path>();
// restart and verify only app resource is marked in-progress
restartStateStore();
RecoveredLocalizationState state = stateStore.loadLocalizationState();
LocalResourceTrackerState pubts = state.getPublicTrackerState();
completedResources = loadCompletedResources(
pubts.getCompletedResourcesIterator());
startedResources = loadStartedResources(
pubts.getStartedResourcesIterator());
assertTrue(completedResources.isEmpty());
assertTrue(startedResources.isEmpty());
Map<String, RecoveredUserResources> userResources =
loadUserResources(state.getIterator());
assertEquals(1, userResources.size());
RecoveredUserResources rur = userResources.get(user);
LocalResourceTrackerState privts = rur.getPrivateTrackerState();
assertNotNull(privts);
completedResources = loadCompletedResources(
privts.getCompletedResourcesIterator());
startedResources = loadStartedResources(
privts.getStartedResourcesIterator());
assertTrue(completedResources.isEmpty());
assertTrue(startedResources.isEmpty());
assertEquals(1, rur.getAppTrackerStates().size());
LocalResourceTrackerState appts = rur.getAppTrackerStates().get(appId);
assertNotNull(appts);
completedResources = loadCompletedResources(
appts.getCompletedResourcesIterator());
startedResources = loadStartedResources(
appts.getStartedResourcesIterator());
assertTrue(completedResources.isEmpty());
assertEquals(1, startedResources.size());
assertEquals(appRsrcLocalPath,
startedResources.get(appRsrcProto));
}
@Test
public void testStartResourceLocalizationForPublicResources()
throws IOException {
Path pubRsrcPath1 = new Path("hdfs://some/public/resource1");
LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource
.newInstance(
URL.fromPath(pubRsrcPath1),
LocalResourceType.FILE, LocalResourceVisibility.PUBLIC,
789L, 135L);
LocalResourceProto pubRsrcProto1 = rsrcPb.getProto();
Path pubRsrcLocalPath1 = new Path("/some/local/dir/for/pubrsrc1");
stateStore.startResourceLocalization(null, null, pubRsrcProto1,
pubRsrcLocalPath1);
Path pubRsrcPath2 = new Path("hdfs://some/public/resource2");
rsrcPb = (LocalResourcePBImpl) LocalResource.newInstance(
URL.fromPath(pubRsrcPath2),
LocalResourceType.FILE, LocalResourceVisibility.PUBLIC,
789L, 135L);
LocalResourceProto pubRsrcProto2 = rsrcPb.getProto();
Path pubRsrcLocalPath2 = new Path("/some/local/dir/for/pubrsrc2");
stateStore.startResourceLocalization(null, null, pubRsrcProto2,
pubRsrcLocalPath2);
// restart and verify resources are marked in-progress
restartStateStore();
RecoveredLocalizationState state = stateStore.loadLocalizationState();
LocalResourceTrackerState pubts = state.getPublicTrackerState();
List<LocalizedResourceProto> completedResources = loadCompletedResources(
pubts.getCompletedResourcesIterator());
Map<LocalResourceProto, Path> startedResources = loadStartedResources(
pubts.getStartedResourcesIterator());
assertTrue(completedResources.isEmpty());
assertEquals(2, startedResources.size());
assertEquals(pubRsrcLocalPath1,
startedResources.get(pubRsrcProto1));
assertEquals(pubRsrcLocalPath2,
startedResources.get(pubRsrcProto2));
Map<String, RecoveredUserResources> userResources =
loadUserResources(state.getIterator());
assertEquals(0, userResources.size());
}
@Test
public void testStartResourceLocalizationForPrivateResource()
throws IOException {
Path privRsrcPath = new Path("hdfs://some/private/resource");
LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource
.newInstance(
URL.fromPath(privRsrcPath),
LocalResourceType.PATTERN, LocalResourceVisibility.PRIVATE,
789L, 680L, "*pattern*");
LocalResourceProto privRsrcProto = rsrcPb.getProto();
Path privRsrcLocalPath = new Path("/some/local/dir/for/privrsrc");
String user = "somebody";
stateStore.startResourceLocalization(user, null, privRsrcProto,
privRsrcLocalPath);
// restart and verify resources are marked in-progress
restartStateStore();
RecoveredLocalizationState state = stateStore.loadLocalizationState();
Map<String, RecoveredUserResources> userResources =
loadUserResources(state.getIterator());
assertEquals(1, userResources.size());
RecoveredUserResources rur = userResources.get(user);
LocalResourceTrackerState privts = rur.getPrivateTrackerState();
assertNotNull(privts);
List<LocalizedResourceProto> completedResources = loadCompletedResources(
privts.getCompletedResourcesIterator());
Map<LocalResourceProto, Path> startedResources = loadStartedResources(
privts.getStartedResourcesIterator());
assertTrue(completedResources.isEmpty());
assertEquals(1, startedResources.size());
assertEquals(privRsrcLocalPath,
startedResources.get(privRsrcProto));
assertEquals(0, rur.getAppTrackerStates().size());
}
@Test
public void testFinishResourceLocalizationForApplicationResource()
throws IOException {
String user = "somebody";
ApplicationId appId = ApplicationId.newInstance(1, 1);
// start and finish a local resource for an application
Path appRsrcPath = new Path("hdfs://some/app/resource");
LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl)
LocalResource.newInstance(
URL.fromPath(appRsrcPath),
LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION,
123L, 456L);
LocalResourceProto appRsrcProto = rsrcPb.getProto();
Path appRsrcLocalPath = new Path("/some/local/dir/for/apprsrc");
stateStore.startResourceLocalization(user, appId, appRsrcProto,
appRsrcLocalPath);
LocalizedResourceProto appLocalizedProto =
LocalizedResourceProto.newBuilder()
.setResource(appRsrcProto)
.setLocalPath(appRsrcLocalPath.toString())
.setSize(1234567L)
.build();
stateStore.finishResourceLocalization(user, appId, appLocalizedProto);
List<LocalizedResourceProto> completedResources =
new ArrayList<LocalizedResourceProto>();
Map<LocalResourceProto, Path> startedResources =
new HashMap<LocalResourceProto, Path>();
// restart and verify only app resource is completed
restartStateStore();
RecoveredLocalizationState state = stateStore.loadLocalizationState();
LocalResourceTrackerState pubts = state.getPublicTrackerState();
completedResources = loadCompletedResources(
pubts.getCompletedResourcesIterator());
startedResources = loadStartedResources(
pubts.getStartedResourcesIterator());
assertTrue(completedResources.isEmpty());
assertTrue(startedResources.isEmpty());
Map<String, RecoveredUserResources> userResources =
loadUserResources(state.getIterator());
assertEquals(1, userResources.size());
RecoveredUserResources rur = userResources.get(user);
LocalResourceTrackerState privts = rur.getPrivateTrackerState();
assertNotNull(privts);
completedResources = loadCompletedResources(
privts.getCompletedResourcesIterator());
startedResources = loadStartedResources(
privts.getStartedResourcesIterator());
assertTrue(completedResources.isEmpty());
assertTrue(startedResources.isEmpty());
assertEquals(1, rur.getAppTrackerStates().size());
LocalResourceTrackerState appts = rur.getAppTrackerStates().get(appId);
assertNotNull(appts);
completedResources = loadCompletedResources(
appts.getCompletedResourcesIterator());
startedResources = loadStartedResources(
appts.getStartedResourcesIterator());
assertTrue(startedResources.isEmpty());
assertEquals(1, completedResources.size());
assertEquals(appLocalizedProto,
completedResources.iterator().next());
}
@Test
public void testFinishResourceLocalizationForPublicResources()
throws IOException {
Path pubRsrcPath1 = new Path("hdfs://some/public/resource1");
LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource
.newInstance(
URL.fromPath(pubRsrcPath1),
LocalResourceType.FILE, LocalResourceVisibility.PUBLIC,
789L, 135L);
LocalResourceProto pubRsrcProto1 = rsrcPb.getProto();
Path pubRsrcLocalPath1 = new Path("/some/local/dir/for/pubrsrc1");
stateStore.startResourceLocalization(null, null, pubRsrcProto1,
pubRsrcLocalPath1);
Path pubRsrcPath2 = new Path("hdfs://some/public/resource2");
rsrcPb = (LocalResourcePBImpl) LocalResource.newInstance(
URL.fromPath(pubRsrcPath2),
LocalResourceType.FILE, LocalResourceVisibility.PUBLIC,
789L, 135L);
LocalResourceProto pubRsrcProto2 = rsrcPb.getProto();
Path pubRsrcLocalPath2 = new Path("/some/local/dir/for/pubrsrc2");
stateStore.startResourceLocalization(null, null, pubRsrcProto2,
pubRsrcLocalPath2);
// finish some of the resources
LocalizedResourceProto pubLocalizedProto1 =
LocalizedResourceProto.newBuilder()
.setResource(pubRsrcProto1)
.setLocalPath(pubRsrcLocalPath1.toString())
.setSize(pubRsrcProto1.getSize())
.build();
stateStore.finishResourceLocalization(null, null, pubLocalizedProto1);
// restart and verify state
restartStateStore();
RecoveredLocalizationState state = stateStore.loadLocalizationState();
LocalResourceTrackerState pubts = state.getPublicTrackerState();
List<LocalizedResourceProto> completedResources = loadCompletedResources(
pubts.getCompletedResourcesIterator());
Map<LocalResourceProto, Path> startedResources = loadStartedResources(
pubts.getStartedResourcesIterator());
assertEquals(1, completedResources.size());
assertEquals(pubLocalizedProto1,
completedResources.iterator().next());
assertEquals(1, startedResources.size());
assertEquals(pubRsrcLocalPath2,
startedResources.get(pubRsrcProto2));
Map<String, RecoveredUserResources> userResources =
loadUserResources(state.getIterator());
assertEquals(0, userResources.size());
}
@Test
public void testFinishResourceLocalizationForPrivateResource()
throws IOException {
String user = "somebody";
ApplicationId appId = ApplicationId.newInstance(1, 1);
Path privRsrcPath = new Path("hdfs://some/private/resource");
LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource
.newInstance(
URL.fromPath(privRsrcPath),
LocalResourceType.PATTERN, LocalResourceVisibility.PRIVATE,
789L, 680L, "*pattern*");
LocalResourceProto privRsrcProto = rsrcPb.getProto();
Path privRsrcLocalPath = new Path("/some/local/dir/for/privrsrc");
stateStore.startResourceLocalization(user, null, privRsrcProto,
privRsrcLocalPath);
LocalizedResourceProto privLocalizedProto =
LocalizedResourceProto.newBuilder()
.setResource(privRsrcProto)
.setLocalPath(privRsrcLocalPath.toString())
.setSize(privRsrcProto.getSize())
.build();
stateStore.finishResourceLocalization(user, null, privLocalizedProto);
// restart and verify state
restartStateStore();
RecoveredLocalizationState state = stateStore.loadLocalizationState();
LocalResourceTrackerState pubts = state.getPublicTrackerState();
List<LocalizedResourceProto> completedResources = loadCompletedResources(
pubts.getCompletedResourcesIterator());
Map<LocalResourceProto, Path> startedResources = loadStartedResources(
pubts.getStartedResourcesIterator());
assertEquals(0, completedResources.size());
assertEquals(0, startedResources.size());
Map<String, RecoveredUserResources> userResources =
loadUserResources(state.getIterator());
assertEquals(1, userResources.size());
RecoveredUserResources rur = userResources.get(user);
LocalResourceTrackerState privts = rur.getPrivateTrackerState();
assertNotNull(privts);
completedResources = loadCompletedResources(
privts.getCompletedResourcesIterator());
startedResources = loadStartedResources(
privts.getStartedResourcesIterator());
assertEquals(1, completedResources.size());
assertEquals(privLocalizedProto,
completedResources.iterator().next());
assertTrue(startedResources.isEmpty());
assertEquals(0, rur.getAppTrackerStates().size());
LocalResourceTrackerState appts = rur.getAppTrackerStates().get(appId);
assertNull(appts);
assertTrue(startedResources.isEmpty());
assertEquals(1, completedResources.size());
}
@Test
public void testRemoveLocalizedResourceForApplicationResource()
throws IOException {
String user = "somebody";
ApplicationId appId = ApplicationId.newInstance(1, 1);
// go through the complete lifecycle for an application local resource
Path appRsrcPath = new Path("hdfs://some/app/resource");
LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl)
LocalResource.newInstance(
URL.fromPath(appRsrcPath),
LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION,
123L, 456L);
LocalResourceProto appRsrcProto = rsrcPb.getProto();
Path appRsrcLocalPath = new Path("/some/local/dir/for/apprsrc");
stateStore.startResourceLocalization(user, appId, appRsrcProto,
appRsrcLocalPath);
LocalizedResourceProto appLocalizedProto =
LocalizedResourceProto.newBuilder()
.setResource(appRsrcProto)
.setLocalPath(appRsrcLocalPath.toString())
.setSize(1234567L)
.build();
stateStore.finishResourceLocalization(user, appId, appLocalizedProto);
stateStore.removeLocalizedResource(user, appId, appRsrcLocalPath);
restartStateStore();
verifyEmptyState();
// remove an app resource that didn't finish
stateStore.startResourceLocalization(user, appId, appRsrcProto,
appRsrcLocalPath);
stateStore.removeLocalizedResource(user, appId, appRsrcLocalPath);
restartStateStore();
verifyEmptyState();
}
@Test
public void testRemoveLocalizedResourceForPublicResources()
throws IOException {
// add public resources and remove some
Path pubRsrcPath1 = new Path("hdfs://some/public/resource1");
LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource
.newInstance(
URL.fromPath(pubRsrcPath1),
LocalResourceType.FILE, LocalResourceVisibility.PUBLIC,
789L, 135L);
LocalResourceProto pubRsrcProto1 = rsrcPb.getProto();
Path pubRsrcLocalPath1 = new Path("/some/local/dir/for/pubrsrc1");
stateStore.startResourceLocalization(null, null, pubRsrcProto1,
pubRsrcLocalPath1);
LocalizedResourceProto pubLocalizedProto1 =
LocalizedResourceProto.newBuilder()
.setResource(pubRsrcProto1)
.setLocalPath(pubRsrcLocalPath1.toString())
.setSize(789L)
.build();
stateStore.finishResourceLocalization(null, null, pubLocalizedProto1);
Path pubRsrcPath2 = new Path("hdfs://some/public/resource2");
rsrcPb = (LocalResourcePBImpl) LocalResource.newInstance(
URL.fromPath(pubRsrcPath2),
LocalResourceType.FILE, LocalResourceVisibility.PUBLIC,
789L, 135L);
LocalResourceProto pubRsrcProto2 = rsrcPb.getProto();
Path pubRsrcLocalPath2 = new Path("/some/local/dir/for/pubrsrc2");
stateStore.startResourceLocalization(null, null, pubRsrcProto2,
pubRsrcLocalPath2);
LocalizedResourceProto pubLocalizedProto2 =
LocalizedResourceProto.newBuilder()
.setResource(pubRsrcProto2)
.setLocalPath(pubRsrcLocalPath2.toString())
.setSize(7654321L)
.build();
stateStore.finishResourceLocalization(null, null, pubLocalizedProto2);
stateStore.removeLocalizedResource(null, null, pubRsrcLocalPath2);
// restart and verify state
restartStateStore();
RecoveredLocalizationState state = stateStore.loadLocalizationState();
LocalResourceTrackerState pubts = state.getPublicTrackerState();
List<LocalizedResourceProto> completedResources =
loadCompletedResources(pubts.getCompletedResourcesIterator());
Map<LocalResourceProto, Path> startedResources =
loadStartedResources(pubts.getStartedResourcesIterator());
assertTrue(startedResources.isEmpty());
assertEquals(1, completedResources.size());
assertEquals(pubLocalizedProto1,
completedResources.iterator().next());
Map<String, RecoveredUserResources> userResources =
loadUserResources(state.getIterator());
assertTrue(userResources.isEmpty());
}
@Test
public void testRemoveLocalizedResourceForPrivateResource()
throws IOException {
String user = "somebody";
Path privRsrcPath = new Path("hdfs://some/private/resource");
LocalResourcePBImpl rsrcPb = (LocalResourcePBImpl) LocalResource
.newInstance(
URL.fromPath(privRsrcPath),
LocalResourceType.PATTERN, LocalResourceVisibility.PRIVATE,
789L, 680L, "*pattern*");
LocalResourceProto privRsrcProto = rsrcPb.getProto();
Path privRsrcLocalPath = new Path("/some/local/dir/for/privrsrc");
stateStore.startResourceLocalization(user, null, privRsrcProto,
privRsrcLocalPath);
stateStore.removeLocalizedResource(user, null, privRsrcLocalPath);
// restart and verify state
restartStateStore();
RecoveredLocalizationState state = stateStore.loadLocalizationState();
LocalResourceTrackerState pubts = state.getPublicTrackerState();
List<LocalizedResourceProto> completedResources =
loadCompletedResources(pubts.getCompletedResourcesIterator());
Map<LocalResourceProto, Path> startedResources =
loadStartedResources(pubts.getStartedResourcesIterator());
assertTrue(startedResources.isEmpty());
assertEquals(0, completedResources.size());
Map<String, RecoveredUserResources> userResources =
loadUserResources(state.getIterator());
assertTrue(userResources.isEmpty());
}
@Test
public void testDeletionTaskStorage() throws IOException {
// test empty when no state
RecoveredDeletionServiceState state =
stateStore.loadDeletionServiceState();
List<DeletionServiceDeleteTaskProto> deleteTaskProtos =
loadDeletionTaskProtos(state.getIterator());
assertTrue(deleteTaskProtos.isEmpty());
// store a deletion task and verify recovered
DeletionServiceDeleteTaskProto proto =
DeletionServiceDeleteTaskProto.newBuilder()
.setId(7)
.setUser("someuser")
.setSubdir("some/subdir")
.addBasedirs("some/dir/path")
.addBasedirs("some/other/dir/path")
.setDeletionTime(123456L)
.addSuccessorIds(8)
.addSuccessorIds(9)
.build();
stateStore.storeDeletionTask(proto.getId(), proto);
restartStateStore();
state = stateStore.loadDeletionServiceState();
deleteTaskProtos = loadDeletionTaskProtos(state.getIterator());
assertEquals(1, deleteTaskProtos.size());
assertEquals(proto, deleteTaskProtos.get(0));
// store another deletion task
DeletionServiceDeleteTaskProto proto2 =
DeletionServiceDeleteTaskProto.newBuilder()
.setId(8)
.setUser("user2")
.setSubdir("subdir2")
.setDeletionTime(789L)
.build();
stateStore.storeDeletionTask(proto2.getId(), proto2);
restartStateStore();
state = stateStore.loadDeletionServiceState();
deleteTaskProtos = loadDeletionTaskProtos(state.getIterator());
assertEquals(2, deleteTaskProtos.size());
assertTrue(deleteTaskProtos.contains(proto));
assertTrue(deleteTaskProtos.contains(proto2));
// delete a task and verify gone after recovery
stateStore.removeDeletionTask(proto2.getId());
restartStateStore();
state = stateStore.loadDeletionServiceState();
deleteTaskProtos = loadDeletionTaskProtos(state.getIterator());
assertEquals(1, deleteTaskProtos.size());
assertEquals(proto, deleteTaskProtos.get(0));
// delete the last task and verify none left
stateStore.removeDeletionTask(proto.getId());
restartStateStore();
state = stateStore.loadDeletionServiceState();
deleteTaskProtos = loadDeletionTaskProtos(state.getIterator());
assertTrue(deleteTaskProtos.isEmpty()); }
@Test
public void testNMTokenStorage() throws IOException {
// test empty when no state
RecoveredNMTokensState state = stateStore.loadNMTokensState();
Map<ApplicationAttemptId, MasterKey> loadedAppKeys =
loadNMTokens(state.getIterator());
assertNull(state.getCurrentMasterKey());
assertNull(state.getPreviousMasterKey());
assertTrue(loadedAppKeys.isEmpty());
// store a master key and verify recovered
NMTokenSecretManagerForTest secretMgr = new NMTokenSecretManagerForTest();
MasterKey currentKey = secretMgr.generateKey();
stateStore.storeNMTokenCurrentMasterKey(currentKey);
restartStateStore();
state = stateStore.loadNMTokensState();
loadedAppKeys = loadNMTokens(state.getIterator());
assertEquals(currentKey, state.getCurrentMasterKey());
assertNull(state.getPreviousMasterKey());
assertTrue(loadedAppKeys.isEmpty());
// store a previous key and verify recovered
MasterKey prevKey = secretMgr.generateKey();
stateStore.storeNMTokenPreviousMasterKey(prevKey);
restartStateStore();
state = stateStore.loadNMTokensState();
loadedAppKeys = loadNMTokens(state.getIterator());
assertEquals(currentKey, state.getCurrentMasterKey());
assertEquals(prevKey, state.getPreviousMasterKey());
assertTrue(loadedAppKeys.isEmpty());
// store a few application keys and verify recovered
ApplicationAttemptId attempt1 = ApplicationAttemptId.newInstance(
ApplicationId.newInstance(1, 1), 1);
MasterKey attemptKey1 = secretMgr.generateKey();
stateStore.storeNMTokenApplicationMasterKey(attempt1, attemptKey1);
ApplicationAttemptId attempt2 = ApplicationAttemptId.newInstance(
ApplicationId.newInstance(2, 3), 4);
MasterKey attemptKey2 = secretMgr.generateKey();
stateStore.storeNMTokenApplicationMasterKey(attempt2, attemptKey2);
restartStateStore();
state = stateStore.loadNMTokensState();
loadedAppKeys = loadNMTokens(state.getIterator());
assertEquals(currentKey, state.getCurrentMasterKey());
assertEquals(prevKey, state.getPreviousMasterKey());
assertEquals(2, loadedAppKeys.size());
assertEquals(attemptKey1, loadedAppKeys.get(attempt1));
assertEquals(attemptKey2, loadedAppKeys.get(attempt2));
// add/update/remove keys and verify recovered
ApplicationAttemptId attempt3 = ApplicationAttemptId.newInstance(
ApplicationId.newInstance(5, 6), 7);
MasterKey attemptKey3 = secretMgr.generateKey();
stateStore.storeNMTokenApplicationMasterKey(attempt3, attemptKey3);
stateStore.removeNMTokenApplicationMasterKey(attempt1);
attemptKey2 = prevKey;
stateStore.storeNMTokenApplicationMasterKey(attempt2, attemptKey2);
prevKey = currentKey;
stateStore.storeNMTokenPreviousMasterKey(prevKey);
currentKey = secretMgr.generateKey();
stateStore.storeNMTokenCurrentMasterKey(currentKey);
restartStateStore();
state = stateStore.loadNMTokensState();
loadedAppKeys = loadNMTokens(state.getIterator());
assertEquals(currentKey, state.getCurrentMasterKey());
assertEquals(prevKey, state.getPreviousMasterKey());
assertEquals(2, loadedAppKeys.size());
assertNull(loadedAppKeys.get(attempt1));
assertEquals(attemptKey2, loadedAppKeys.get(attempt2));
assertEquals(attemptKey3, loadedAppKeys.get(attempt3));
}
@Test
public void testContainerTokenStorage() throws IOException {
// test empty when no state
RecoveredContainerTokensState state =
stateStore.loadContainerTokensState();
Map<ContainerId, Long> loadedActiveTokens = loadContainerTokens(state.it);
assertNull(state.getCurrentMasterKey());
assertNull(state.getPreviousMasterKey());
assertTrue(loadedActiveTokens.isEmpty());
// store a master key and verify recovered
ContainerTokenKeyGeneratorForTest keygen =
new ContainerTokenKeyGeneratorForTest(new YarnConfiguration());
MasterKey currentKey = keygen.generateKey();
stateStore.storeContainerTokenCurrentMasterKey(currentKey);
restartStateStore();
state = stateStore.loadContainerTokensState();
loadedActiveTokens = loadContainerTokens(state.it);
assertEquals(currentKey, state.getCurrentMasterKey());
assertNull(state.getPreviousMasterKey());
assertTrue(loadedActiveTokens.isEmpty());
// store a previous key and verify recovered
MasterKey prevKey = keygen.generateKey();
stateStore.storeContainerTokenPreviousMasterKey(prevKey);
restartStateStore();
state = stateStore.loadContainerTokensState();
loadedActiveTokens = loadContainerTokens(state.it);
assertEquals(currentKey, state.getCurrentMasterKey());
assertEquals(prevKey, state.getPreviousMasterKey());
assertTrue(loadedActiveTokens.isEmpty());
// store a few container tokens and verify recovered
ContainerId cid1 = BuilderUtils.newContainerId(1, 1, 1, 1);
Long expTime1 = 1234567890L;
ContainerId cid2 = BuilderUtils.newContainerId(2, 2, 2, 2);
Long expTime2 = 9876543210L;
stateStore.storeContainerToken(cid1, expTime1);
stateStore.storeContainerToken(cid2, expTime2);
restartStateStore();
state = stateStore.loadContainerTokensState();
loadedActiveTokens = loadContainerTokens(state.it);
assertEquals(currentKey, state.getCurrentMasterKey());
assertEquals(prevKey, state.getPreviousMasterKey());
assertEquals(2, loadedActiveTokens.size());
assertEquals(expTime1, loadedActiveTokens.get(cid1));
assertEquals(expTime2, loadedActiveTokens.get(cid2));
// add/update/remove tokens and verify recovered
ContainerId cid3 = BuilderUtils.newContainerId(3, 3, 3, 3);
Long expTime3 = 135798642L;
stateStore.storeContainerToken(cid3, expTime3);
stateStore.removeContainerToken(cid1);
expTime2 += 246897531L;
stateStore.storeContainerToken(cid2, expTime2);
prevKey = currentKey;
stateStore.storeContainerTokenPreviousMasterKey(prevKey);
currentKey = keygen.generateKey();
stateStore.storeContainerTokenCurrentMasterKey(currentKey);
restartStateStore();
state = stateStore.loadContainerTokensState();
loadedActiveTokens = loadContainerTokens(state.it);
assertEquals(currentKey, state.getCurrentMasterKey());
assertEquals(prevKey, state.getPreviousMasterKey());
assertEquals(2, loadedActiveTokens.size());
assertNull(loadedActiveTokens.get(cid1));
assertEquals(expTime2, loadedActiveTokens.get(cid2));
assertEquals(expTime3, loadedActiveTokens.get(cid3));
}
@Test
public void testLogDeleterStorage() throws IOException {
// test empty when no state
RecoveredLogDeleterState state = stateStore.loadLogDeleterState();
assertTrue(state.getLogDeleterMap().isEmpty());
// store log deleter state
final ApplicationId appId1 = ApplicationId.newInstance(1, 1);
LogDeleterProto proto1 = LogDeleterProto.newBuilder()
.setUser("user1")
.setDeletionTime(1234)
.build();
stateStore.storeLogDeleter(appId1, proto1);
// restart state store and verify recovered
restartStateStore();
state = stateStore.loadLogDeleterState();
assertEquals(1, state.getLogDeleterMap().size());
assertEquals(proto1, state.getLogDeleterMap().get(appId1));
// store another log deleter
final ApplicationId appId2 = ApplicationId.newInstance(2, 2);
LogDeleterProto proto2 = LogDeleterProto.newBuilder()
.setUser("user2")
.setDeletionTime(5678)
.build();
stateStore.storeLogDeleter(appId2, proto2);
// restart state store and verify recovered
restartStateStore();
state = stateStore.loadLogDeleterState();
assertEquals(2, state.getLogDeleterMap().size());
assertEquals(proto1, state.getLogDeleterMap().get(appId1));
assertEquals(proto2, state.getLogDeleterMap().get(appId2));
// remove a deleter and verify removed after restart and recovery
stateStore.removeLogDeleter(appId1);
restartStateStore();
state = stateStore.loadLogDeleterState();
assertEquals(1, state.getLogDeleterMap().size());
assertEquals(proto2, state.getLogDeleterMap().get(appId2));
// remove last deleter and verify empty after restart and recovery
stateStore.removeLogDeleter(appId2);
restartStateStore();
state = stateStore.loadLogDeleterState();
assertTrue(state.getLogDeleterMap().isEmpty());
}
@Test
public void testCompactionCycle() throws IOException {
final DB mockdb = mock(DB.class);
conf.setInt(YarnConfiguration.NM_RECOVERY_COMPACTION_INTERVAL_SECS, 1);
NMLeveldbStateStoreService store = new NMLeveldbStateStoreService() {
@Override
protected void checkVersion() {}
@Override
protected DB openDatabase(Configuration conf) {
return mockdb;
}
};
store.init(conf);
store.start();
verify(mockdb, timeout(10000).atLeastOnce()).compactRange(
(byte[]) isNull(), (byte[]) isNull());
store.close();
}
@Test
public void testUnexpectedKeyDoesntThrowException() throws IOException {
// test empty when no state
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertTrue(recoveredContainers.isEmpty());
ApplicationId appId = ApplicationId.newInstance(1234, 3);
ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(appId,
4);
ContainerId containerId = ContainerId.newContainerId(appAttemptId, 5);
StartContainerRequest startContainerRequest = storeMockContainer(
containerId);
// add a invalid key
byte[] invalidKey = ("ContainerManager/containers/"
+ containerId.toString() + "/invalidKey1234").getBytes();
stateStore.getDB().put(invalidKey, new byte[1]);
restartStateStore();
recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertEquals(1, recoveredContainers.size());
RecoveredContainerState rcs = recoveredContainers.get(0);
assertEquals(RecoveredContainerStatus.REQUESTED, rcs.getStatus());
assertEquals(ContainerExitStatus.INVALID, rcs.getExitCode());
assertEquals(false, rcs.getKilled());
assertEquals(startContainerRequest, rcs.getStartRequest());
assertTrue(rcs.getDiagnostics().isEmpty());
assertEquals(RecoveredContainerType.KILL, rcs.getRecoveryType());
// assert unknown keys are cleaned up finally
assertNotNull(stateStore.getDB().get(invalidKey));
stateStore.removeContainer(containerId);
assertNull(stateStore.getDB().get(invalidKey));
}
@Test
public void testAMRMProxyStorage() throws IOException {
RecoveredAMRMProxyState state = stateStore.loadAMRMProxyState();
assertThat(state.getCurrentMasterKey()).isNull();
assertThat(state.getNextMasterKey()).isNull();
assertThat(state.getAppContexts()).isEmpty();
ApplicationId appId1 = ApplicationId.newInstance(1, 1);
ApplicationId appId2 = ApplicationId.newInstance(1, 2);
ApplicationAttemptId attemptId1 =
ApplicationAttemptId.newInstance(appId1, 1);
ApplicationAttemptId attemptId2 =
ApplicationAttemptId.newInstance(appId2, 2);
String key1 = "key1";
String key2 = "key2";
byte[] data1 = "data1".getBytes();
byte[] data2 = "data2".getBytes();
AMRMProxyTokenSecretManager secretManager =
new AMRMProxyTokenSecretManager(stateStore);
secretManager.init(conf);
// Generate currentMasterKey
secretManager.start();
try {
// Add two applications, each with two data entries
stateStore.storeAMRMProxyAppContextEntry(attemptId1, key1, data1);
stateStore.storeAMRMProxyAppContextEntry(attemptId2, key1, data1);
stateStore.storeAMRMProxyAppContextEntry(attemptId1, key2, data2);
stateStore.storeAMRMProxyAppContextEntry(attemptId2, key2, data2);
// restart state store and verify recovered
restartStateStore();
secretManager.setNMStateStoreService(stateStore);
state = stateStore.loadAMRMProxyState();
assertEquals(state.getCurrentMasterKey(),
secretManager.getCurrentMasterKeyData().getMasterKey());
assertThat(state.getNextMasterKey()).isNull();
assertThat(state.getAppContexts()).hasSize(2);
// app1
Map<String, byte[]> map = state.getAppContexts().get(attemptId1);
assertNotEquals(map, null);
assertThat(map).hasSize(2);
assertTrue(Arrays.equals(map.get(key1), data1));
assertTrue(Arrays.equals(map.get(key2), data2));
// app2
map = state.getAppContexts().get(attemptId2);
assertNotEquals(map, null);
assertThat(map).hasSize(2);
assertTrue(Arrays.equals(map.get(key1), data1));
assertTrue(Arrays.equals(map.get(key2), data2));
// Generate next master key and remove one entry of app2
secretManager.rollMasterKey();
stateStore.removeAMRMProxyAppContextEntry(attemptId2, key1);
// restart state store and verify recovered
restartStateStore();
secretManager.setNMStateStoreService(stateStore);
state = stateStore.loadAMRMProxyState();
assertEquals(state.getCurrentMasterKey(),
secretManager.getCurrentMasterKeyData().getMasterKey());
assertEquals(state.getNextMasterKey(),
secretManager.getNextMasterKeyData().getMasterKey());
assertEquals(state.getAppContexts().size(), 2);
// app1
map = state.getAppContexts().get(attemptId1);
assertThat(map).isNotNull();
assertThat(map).hasSize(2);
assertTrue(Arrays.equals(map.get(key1), data1));
assertTrue(Arrays.equals(map.get(key2), data2));
// app2
map = state.getAppContexts().get(attemptId2);
assertThat(map).isNotNull();
assertThat(map).hasSize(1);
assertTrue(Arrays.equals(map.get(key2), data2));
// Activate next master key and remove all entries of app1
secretManager.activateNextMasterKey();
stateStore.removeAMRMProxyAppContext(attemptId1);
// restart state store and verify recovered
restartStateStore();
secretManager.setNMStateStoreService(stateStore);
state = stateStore.loadAMRMProxyState();
assertEquals(state.getCurrentMasterKey(),
secretManager.getCurrentMasterKeyData().getMasterKey());
assertThat(state.getNextMasterKey()).isNull();
assertThat(state.getAppContexts()).hasSize(1);
// app2 only
map = state.getAppContexts().get(attemptId2);
assertThat(map).isNotNull();
assertThat(map).hasSize(1);
assertTrue(Arrays.equals(map.get(key2), data2));
} finally {
secretManager.stop();
}
}
@Test
public void testStateStoreForResourceMapping() throws IOException {
// test that stateStore is initially empty
List<RecoveredContainerState> recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertTrue(recoveredContainers.isEmpty());
ApplicationId appId = ApplicationId.newInstance(1234, 3);
ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(appId,
4);
ContainerId containerId = ContainerId.newContainerId(appAttemptId, 5);
storeMockContainer(containerId);
Container container = mock(Container.class);
when(container.getContainerId()).thenReturn(containerId);
ResourceMappings resourceMappings = new ResourceMappings();
when(container.getResourceMappings()).thenReturn(resourceMappings);
stateStore.storeAssignedResources(container, "gpu",
Arrays.asList(new GpuDevice(1, 1), new GpuDevice(2, 2),
new GpuDevice(3, 3)));
// This will overwrite the above
List<Serializable> gpuRes1 = Arrays.asList(
new GpuDevice(1, 1), new GpuDevice(2, 2), new GpuDevice(4, 4));
stateStore.storeAssignedResources(container, "gpu", gpuRes1);
List<Serializable> fpgaRes = Arrays.asList(
new FpgaDevice("testType", 3, 3, "testIPID"),
new FpgaDevice("testType", 4, 4, "testIPID"),
new FpgaDevice("testType", 5, 5, "testIPID"),
new FpgaDevice("testType", 6, 6, "testIPID"));
stateStore.storeAssignedResources(container, "fpga", fpgaRes);
List<Serializable> numaRes = Arrays.asList(
new NumaResourceAllocation("testmemNodeId", 2048, "testCpuNodeId", 10));
stateStore.storeAssignedResources(container, "numa", numaRes);
restartStateStore();
recoveredContainers =
loadContainersState(stateStore.getContainerStateIterator());
assertEquals(1, recoveredContainers.size());
RecoveredContainerState rcs = recoveredContainers.get(0);
List<Serializable> resources = rcs.getResourceMappings()
.getAssignedResources("gpu");
Assert.assertEquals(gpuRes1, resources);
Assert.assertEquals(gpuRes1, resourceMappings.getAssignedResources("gpu"));
resources = rcs.getResourceMappings().getAssignedResources("fpga");
Assert.assertEquals(fpgaRes, resources);
Assert.assertEquals(fpgaRes, resourceMappings.getAssignedResources("fpga"));
resources = rcs.getResourceMappings().getAssignedResources("numa");
Assert.assertEquals(numaRes, resources);
Assert.assertEquals(numaRes, resourceMappings.getAssignedResources("numa"));
// test removing numa resources from state store
stateStore.releaseAssignedResources(containerId, "numa");
recoveredContainers = loadContainersState(stateStore.getContainerStateIterator());
resourceMappings = recoveredContainers.get(0).getResourceMappings();
assertTrue(resourceMappings.getAssignedResources("numa").isEmpty());
// testing calling deletion of non-existing key doesn't break anything
try {
stateStore.releaseAssignedResources(containerId, "numa");
}catch (RuntimeException e){
Assert.fail("Should not throw exception while deleting non existing key from statestore");
}
}
@Test
public void testStateStoreNodeHealth() throws IOException {
// keep the working DB clean, break a temp DB
DB keepDB = stateStore.getDB();
DB myMocked = mock(DB.class);
stateStore.setDB(myMocked);
ApplicationId appId = ApplicationId.newInstance(1234, 1);
ApplicationAttemptId appAttemptId =
ApplicationAttemptId.newInstance(appId, 1);
DBException toThrow = new DBException();
Mockito.doThrow(toThrow).when(myMocked).
put(any(byte[].class), any(byte[].class));
// write some data
try {
// chosen a simple method could be any of the "void" methods
ContainerId containerId = ContainerId.newContainerId(appAttemptId, 1);
stateStore.storeContainerKilled(containerId);
} catch (IOException ioErr) {
// Cause should be wrapped DBException
assertTrue(ioErr.getCause() instanceof DBException);
// check the store is marked unhealthy
assertFalse("Statestore should have been unhealthy",
stateStore.isHealthy());
return;
} finally {
// restore the working DB
stateStore.setDB(keepDB);
}
Assert.fail("Expected exception not thrown");
}
@Test
public void testEmptyRestartTimes() throws IOException {
List<Long> restartTimes = new ArrayList<>();
ApplicationId appId = ApplicationId.newInstance(1234, 3);
ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(appId,
4);
ContainerId containerId = ContainerId.newContainerId(appAttemptId, 5);
storeMockContainer(containerId);
stateStore.storeContainerRestartTimes(containerId,
restartTimes);
restartStateStore();
RecoveredContainerState rcs =
loadContainersState(stateStore.getContainerStateIterator()).get(0);
List<Long> recoveredRestartTimes = rcs.getRestartTimes();
assertTrue(recoveredRestartTimes.isEmpty());
}
private StartContainerRequest storeMockContainer(ContainerId containerId)
throws IOException {
// create a container request
LocalResource lrsrc = LocalResource.newInstance(
URL.newInstance("hdfs", "somehost", 12345, "/some/path/to/rsrc"),
LocalResourceType.FILE, LocalResourceVisibility.APPLICATION, 123L,
1234567890L);
Map<String, LocalResource> localResources =
new HashMap<String, LocalResource>();
localResources.put("rsrc", lrsrc);
Map<String, String> env = new HashMap<String, String>();
env.put("somevar", "someval");
List<String> containerCmds = new ArrayList<String>();
containerCmds.add("somecmd");
containerCmds.add("somearg");
Map<String, ByteBuffer> serviceData = new HashMap<String, ByteBuffer>();
serviceData.put("someservice",
ByteBuffer.wrap(new byte[] {0x1, 0x2, 0x3}));
ByteBuffer containerTokens = ByteBuffer
.wrap(new byte[] {0x7, 0x8, 0x9, 0xa});
Map<ApplicationAccessType, String> acls =
new HashMap<ApplicationAccessType, String>();
acls.put(ApplicationAccessType.VIEW_APP, "viewuser");
acls.put(ApplicationAccessType.MODIFY_APP, "moduser");
ContainerLaunchContext clc = ContainerLaunchContext.newInstance(
localResources, env, containerCmds,
serviceData, containerTokens, acls);
Resource containerRsrc = Resource.newInstance(1357, 3);
ContainerTokenIdentifier containerTokenId = new ContainerTokenIdentifier(
containerId, "host", "user", containerRsrc, 9876543210L, 42, 2468,
Priority.newInstance(7), 13579);
Token containerToken = Token.newInstance(containerTokenId.getBytes(),
ContainerTokenIdentifier.KIND.toString(), "password".getBytes(),
"tokenservice");
StartContainerRequest containerReq = StartContainerRequest.newInstance(clc,
containerToken);
stateStore.storeContainer(containerId, 0, 0, containerReq);
return containerReq;
}
private static class NMTokenSecretManagerForTest extends
BaseNMTokenSecretManager {
public MasterKey generateKey() {
return createNewMasterKey().getMasterKey();
}
}
private static class ContainerTokenKeyGeneratorForTest extends
BaseContainerTokenSecretManager {
public ContainerTokenKeyGeneratorForTest(Configuration conf) {
super(conf);
}
public MasterKey generateKey() {
return createNewMasterKey().getMasterKey();
}
}
}