You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by xy...@apache.org on 2018/10/18 17:44:55 UTC
[30/50] [abbrv] hadoop git commit: HDDS-662. Introduce
ContainerReplicaState in StorageContainerManager. Contributed by Nanda kumar.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/50715c06/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManager.java
index 7078b8f..42b39f9 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManager.java
@@ -17,17 +17,12 @@
package org.apache.hadoop.hdds.scm.container;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.StorageUnit;
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
-import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
-import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerInfo;
import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
-import org.apache.hadoop.hdds.scm.container.replication.ReplicationRequest;
import org.apache.hadoop.hdds.scm.container.states.ContainerState;
import org.apache.hadoop.hdds.scm.container.states.ContainerStateMap;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
@@ -45,11 +40,8 @@ import org.apache.hadoop.util.Time;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.io.Closeable;
import java.io.IOException;
-import java.util.ArrayList;
import java.util.HashSet;
-import java.util.List;
import java.util.Map;
import java.util.NavigableSet;
import java.util.Set;
@@ -116,7 +108,7 @@ import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes
* TimeOut Delete Container State Machine - if the container creating times out,
* then Container State manager decides to delete the container.
*/
-public class ContainerStateManager implements Closeable {
+public class ContainerStateManager {
private static final Logger LOG =
LoggerFactory.getLogger(ContainerStateManager.class);
@@ -135,11 +127,10 @@ public class ContainerStateManager implements Closeable {
* TODO : Add Container Tags so we know which containers are owned by SCM.
*/
@SuppressWarnings("unchecked")
- public ContainerStateManager(Configuration configuration,
- ContainerManager containerManager, PipelineSelector pipelineSelector) {
+ public ContainerStateManager(final Configuration configuration) {
// Initialize the container state machine.
- Set<HddsProtos.LifeCycleState> finalStates = new HashSet();
+ final Set<HddsProtos.LifeCycleState> finalStates = new HashSet();
// These are the steady states of a container.
finalStates.add(LifeCycleState.OPEN);
@@ -155,22 +146,9 @@ public class ContainerStateManager implements Closeable {
ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT,
StorageUnit.BYTES);
- lastUsedMap = new ConcurrentHashMap<>();
- containerCount = new AtomicLong(0);
- containers = new ContainerStateMap();
- }
-
- /**
- * Return the info of all the containers kept by the in-memory mapping.
- *
- * @return the list of all container info.
- */
- public List<ContainerInfo> getAllContainers() {
- List<ContainerInfo> list = new ArrayList<>();
-
- //No Locking needed since the return value is an immutable map.
- containers.getContainerMap().forEach((key, value) -> list.add(value));
- return list;
+ this.lastUsedMap = new ConcurrentHashMap<>();
+ this.containerCount = new AtomicLong(0);
+ this.containers = new ContainerStateMap();
}
/*
@@ -244,17 +222,15 @@ public class ContainerStateManager implements Closeable {
LifeCycleEvent.CLEANUP);
}
- public void addExistingContainer(ContainerInfo containerInfo)
+ void loadContainer(final ContainerInfo containerInfo)
throws SCMException {
containers.addContainer(containerInfo);
- long containerID = containerInfo.getContainerID();
- if (containerCount.get() < containerID) {
- containerCount.set(containerID);
- }
+ containerCount.set(Long.max(
+ containerInfo.getContainerID(), containerCount.get()));
}
/**
- * allocates a new container based on the type, replication etc.
+ * Allocates a new container based on the type, replication etc.
*
* @param selector -- Pipeline selector class.
* @param type -- Replication type.
@@ -262,25 +238,22 @@ public class ContainerStateManager implements Closeable {
* @return ContainerWithPipeline
* @throws IOException on Failure.
*/
- public ContainerWithPipeline allocateContainer(PipelineSelector selector,
- HddsProtos.ReplicationType type,
- HddsProtos.ReplicationFactor replicationFactor, String owner)
+ ContainerInfo allocateContainer(final PipelineSelector selector,
+ final HddsProtos.ReplicationType type,
+ final HddsProtos.ReplicationFactor replicationFactor, final String owner)
throws IOException {
- Pipeline pipeline = selector.getReplicationPipeline(type,
+ final Pipeline pipeline = selector.getReplicationPipeline(type,
replicationFactor);
Preconditions.checkNotNull(pipeline, "Pipeline type=%s/"
+ "replication=%s couldn't be found for the new container. "
+ "Do you have enough nodes?", type, replicationFactor);
- long containerID = containerCount.incrementAndGet();
- ContainerInfo containerInfo = new ContainerInfo.Builder()
+ final long containerID = containerCount.incrementAndGet();
+ final ContainerInfo containerInfo = new ContainerInfo.Builder()
.setState(HddsProtos.LifeCycleState.ALLOCATED)
.setPipelineID(pipeline.getId())
- // This is bytes allocated for blocks inside container, not the
- // container size
- .setAllocatedBytes(0)
.setUsedBytes(0)
.setNumberOfKeys(0)
.setStateEnterTime(Time.monotonicNow())
@@ -294,35 +267,34 @@ public class ContainerStateManager implements Closeable {
Preconditions.checkNotNull(containerInfo);
containers.addContainer(containerInfo);
LOG.trace("New container allocated: {}", containerInfo);
- return new ContainerWithPipeline(containerInfo, pipeline);
+ return containerInfo;
}
/**
* Update the Container State to the next state.
*
- * @param info - ContainerInfo
+ * @param containerID - ContainerID
* @param event - LifeCycle Event
* @return Updated ContainerInfo.
* @throws SCMException on Failure.
*/
- public ContainerInfo updateContainerState(ContainerInfo
- info, HddsProtos.LifeCycleEvent event) throws SCMException {
- LifeCycleState newState;
+ ContainerInfo updateContainerState(final ContainerID containerID,
+ final HddsProtos.LifeCycleEvent event)
+ throws SCMException, ContainerNotFoundException {
+ final ContainerInfo info = containers.getContainerInfo(containerID);
try {
- newState = this.stateMachine.getNextState(info.getState(), event);
+ final LifeCycleState newState = stateMachine.getNextState(
+ info.getState(), event);
+ containers.updateState(containerID, info.getState(), newState);
+ return containers.getContainerInfo(containerID);
} catch (InvalidStateTransitionException ex) {
String error = String.format("Failed to update container state %s, " +
"reason: invalid state transition from state: %s upon " +
"event: %s.",
- info.getContainerID(), info.getState(), event);
+ containerID, info.getState(), event);
LOG.error(error);
throw new SCMException(error, FAILED_TO_CHANGE_CONTAINER_STATE);
}
-
- // This is a post condition after executing getNextState.
- Preconditions.checkNotNull(newState);
- containers.updateState(info, info.getState(), newState);
- return containers.getContainerInfo(info);
}
/**
@@ -331,10 +303,10 @@ public class ContainerStateManager implements Closeable {
* @return ContainerInfo
* @throws SCMException - on Error.
*/
- public ContainerInfo updateContainerInfo(ContainerInfo info)
- throws SCMException {
+ ContainerInfo updateContainerInfo(final ContainerInfo info)
+ throws ContainerNotFoundException {
containers.updateContainerInfo(info);
- return containers.getContainerInfo(info);
+ return containers.getContainerInfo(info.containerID());
}
/**
@@ -343,11 +315,16 @@ public class ContainerStateManager implements Closeable {
* @param deleteTransactionMap maps containerId to its new
* deleteTransactionID
*/
- public void updateDeleteTransactionId(Map<Long, Long> deleteTransactionMap) {
- for (Map.Entry<Long, Long> entry : deleteTransactionMap.entrySet()) {
- containers.getContainerMap().get(ContainerID.valueof(entry.getKey()))
- .updateDeleteTransactionId(entry.getValue());
- }
+ void updateDeleteTransactionId(
+ final Map<Long, Long> deleteTransactionMap) {
+ deleteTransactionMap.forEach((k, v) -> {
+ try {
+ containers.getContainerInfo(ContainerID.valueof(k))
+ .updateDeleteTransactionId(v);
+ } catch (ContainerNotFoundException e) {
+ LOG.warn("Exception while updating delete transaction id.", e);
+ }
+ });
}
/**
@@ -360,12 +337,12 @@ public class ContainerStateManager implements Closeable {
* @param state - State of the Container-- {Open, Allocated etc.}
* @return ContainerInfo, null if there is no match found.
*/
- public ContainerInfo getMatchingContainer(final long size,
+ ContainerInfo getMatchingContainer(final long size,
String owner, ReplicationType type, ReplicationFactor factor,
LifeCycleState state) {
// Find containers that match the query spec, if no match return null.
- NavigableSet<ContainerID> matchingSet =
+ final NavigableSet<ContainerID> matchingSet =
containers.getMatchingContainerIDs(state, owner, factor, type);
if (matchingSet == null || matchingSet.size() == 0) {
return null;
@@ -373,11 +350,9 @@ public class ContainerStateManager implements Closeable {
// Get the last used container and find container above the last used
// container ID.
- ContainerState key = new ContainerState(owner, type, factor);
- ContainerID lastID = lastUsedMap.get(key);
- if (lastID == null) {
- lastID = matchingSet.first();
- }
+ final ContainerState key = new ContainerState(owner, type, factor);
+ final ContainerID lastID = lastUsedMap
+ .getOrDefault(key, matchingSet.first());
// There is a small issue here. The first time, we will skip the first
// container. But in most cases it will not matter.
@@ -401,32 +376,47 @@ public class ContainerStateManager implements Closeable {
resultSet = matchingSet.headSet(lastID, true);
selectedContainer = findContainerWithSpace(size, resultSet, owner);
}
- // Update the allocated Bytes on this container.
- if (selectedContainer != null) {
- selectedContainer.updateAllocatedBytes(size);
- }
return selectedContainer;
}
- private ContainerInfo findContainerWithSpace(long size,
- NavigableSet<ContainerID> searchSet, String owner) {
- // Get the container with space to meet our request.
- for (ContainerID id : searchSet) {
- ContainerInfo containerInfo = containers.getContainerInfo(id);
- if (containerInfo.getAllocatedBytes() + size <= this.containerSize) {
- containerInfo.updateLastUsedTime();
-
- ContainerState key = new ContainerState(owner,
- containerInfo.getReplicationType(),
- containerInfo.getReplicationFactor());
- lastUsedMap.put(key, containerInfo.containerID());
- return containerInfo;
+ private ContainerInfo findContainerWithSpace(final long size,
+ final NavigableSet<ContainerID> searchSet, final String owner) {
+ try {
+ // Get the container with space to meet our request.
+ for (ContainerID id : searchSet) {
+ final ContainerInfo containerInfo = containers.getContainerInfo(id);
+ if (containerInfo.getUsedBytes() + size <= this.containerSize) {
+ containerInfo.updateLastUsedTime();
+
+ final ContainerState key = new ContainerState(owner,
+ containerInfo.getReplicationType(),
+ containerInfo.getReplicationFactor());
+ lastUsedMap.put(key, containerInfo.containerID());
+ return containerInfo;
+ }
}
+ } catch (ContainerNotFoundException e) {
+ // This should not happen!
+ LOG.warn("Exception while finding container with space", e);
}
return null;
}
+ Set<ContainerID> getAllContainerIDs() {
+ return containers.getAllContainerIDs();
+ }
+
+ /**
+ * Returns Containers by State.
+ *
+ * @param state - State - Open, Closed etc.
+ * @return List of containers by state.
+ */
+ Set<ContainerID> getContainerIDsByState(final LifeCycleState state) {
+ return containers.getContainerIDsByState(state);
+ }
+
/**
* Returns a set of ContainerIDs that match the Container.
*
@@ -436,39 +426,25 @@ public class ContainerStateManager implements Closeable {
* @param state - Current State, like Open, Close etc.
* @return Set of containers that match the specific query parameters.
*/
- public NavigableSet<ContainerID> getMatchingContainerIDs(
- String owner, ReplicationType type, ReplicationFactor factor,
- LifeCycleState state) {
+ NavigableSet<ContainerID> getMatchingContainerIDs(final String owner,
+ final ReplicationType type, final ReplicationFactor factor,
+ final LifeCycleState state) {
return containers.getMatchingContainerIDs(state, owner,
factor, type);
}
/**
- * Returns the containerInfo with pipeline for the given container id.
- * @param selector -- Pipeline selector class.
- * @param containerID id of the container
- * @return ContainerInfo containerInfo
- * @throws IOException
- */
- public ContainerWithPipeline getContainer(PipelineSelector selector,
- ContainerID containerID) {
- ContainerInfo info = containers.getContainerInfo(containerID.getId());
- Pipeline pipeline = selector.getPipeline(info.getPipelineID());
- return new ContainerWithPipeline(info, pipeline);
- }
-
- /**
* Returns the containerInfo for the given container id.
* @param containerID id of the container
* @return ContainerInfo containerInfo
* @throws IOException
*/
- public ContainerInfo getContainer(ContainerID containerID) {
+ ContainerInfo getContainer(final ContainerID containerID)
+ throws ContainerNotFoundException {
return containers.getContainerInfo(containerID);
}
- @Override
- public void close() throws IOException {
+ void close() throws IOException {
}
/**
@@ -478,8 +454,8 @@ public class ContainerStateManager implements Closeable {
* @param containerID
* @return Set<DatanodeDetails>
*/
- public Set<DatanodeDetails> getContainerReplicas(ContainerID containerID)
- throws SCMException {
+ Set<ContainerReplica> getContainerReplicas(
+ final ContainerID containerID) throws ContainerNotFoundException {
return containers.getContainerReplicas(containerID);
}
@@ -487,53 +463,29 @@ public class ContainerStateManager implements Closeable {
* Add a container Replica for given DataNode.
*
* @param containerID
- * @param dn
+ * @param replica
*/
- public void addContainerReplica(ContainerID containerID, DatanodeDetails dn) {
- containers.addContainerReplica(containerID, dn);
+ void updateContainerReplica(final ContainerID containerID,
+ final ContainerReplica replica) throws ContainerNotFoundException {
+ containers.updateContainerReplica(containerID, replica);
}
/**
* Remove a container Replica for given DataNode.
*
* @param containerID
- * @param dn
+ * @param replica
* @return True of dataNode is removed successfully else false.
*/
- public boolean removeContainerReplica(ContainerID containerID,
- DatanodeDetails dn) throws SCMException {
- return containers.removeContainerReplica(containerID, dn);
- }
-
- /**
- * Compare the existing replication number with the expected one.
- */
- public ReplicationRequest checkReplicationState(ContainerID containerID)
- throws SCMException {
- int existingReplicas = getContainerReplicas(containerID).size();
- int expectedReplicas = getContainer(containerID)
- .getReplicationFactor().getNumber();
- if (existingReplicas != expectedReplicas) {
- return new ReplicationRequest(containerID.getId(), existingReplicas,
- expectedReplicas);
- }
- return null;
- }
-
- /**
- * Checks if the container is open.
- */
- public boolean isOpen(ContainerID containerID) {
- Preconditions.checkNotNull(containerID);
- ContainerInfo container = Preconditions
- .checkNotNull(getContainer(containerID),
- "Container can't be found " + containerID);
- return container.isContainerOpen();
+ void removeContainerReplica(final ContainerID containerID,
+ final ContainerReplica replica)
+ throws ContainerNotFoundException, ContainerReplicaNotFoundException {
+ containers.removeContainerReplica(containerID, replica);
}
- @VisibleForTesting
- public ContainerStateMap getContainerStateMap() {
- return containers;
+ void removeContainer(final ContainerID containerID)
+ throws ContainerNotFoundException {
+ containers.removeContainer(containerID);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/50715c06/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java
index 9d6cadb..96ad731 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/SCMContainerManager.java
@@ -24,12 +24,10 @@ import org.apache.hadoop.conf.StorageUnit;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleEvent;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.SCMContainerInfo;
import org.apache.hadoop.hdds.scm.block.PendingDeleteStatusList;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
import org.apache.hadoop.hdds.scm.container.common.helpers.Pipeline;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
-import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerInfo;
import org.apache.hadoop.hdds.scm.container.common.helpers.PipelineID;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
@@ -55,19 +53,23 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
-import java.nio.charset.Charset;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys
.OZONE_SCM_CONTAINER_SIZE_DEFAULT;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys
.OZONE_SCM_CONTAINER_SIZE;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DB_CACHE_SIZE_DEFAULT;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DB_CACHE_SIZE_MB;
import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes
.FAILED_TO_CHANGE_CONTAINER_STATE;
import static org.apache.hadoop.hdds.server.ServerUtils.getOzoneMetaDirPath;
@@ -82,10 +84,7 @@ public class SCMContainerManager implements ContainerManager {
private static final Logger LOG = LoggerFactory.getLogger(SCMContainerManager
.class);
- private final NodeManager nodeManager;
- private final long cacheSize;
private final Lock lock;
- private final Charset encoding = Charset.forName("UTF-8");
private final MetadataStore containerStore;
private final PipelineSelector pipelineSelector;
private final ContainerStateManager containerStateManager;
@@ -100,113 +99,110 @@ public class SCMContainerManager implements ContainerManager {
* @param nodeManager - NodeManager so that we can get the nodes that are
* healthy to place new
* containers.
- * @param cacheSizeMB - Amount of memory reserved for the LSM tree to cache
- * its nodes. This is
* passed to LevelDB and this memory is allocated in Native code space.
* CacheSize is specified
* in MB.
* @throws IOException on Failure.
*/
@SuppressWarnings("unchecked")
- public SCMContainerManager(
- final Configuration conf, final NodeManager nodeManager, final int
- cacheSizeMB, EventPublisher eventPublisher) throws IOException {
- this.nodeManager = nodeManager;
- this.cacheSize = cacheSizeMB;
-
- File metaDir = getOzoneMetaDirPath(conf);
-
- // Write the container name to pipeline mapping.
- File containerDBPath = new File(metaDir, SCM_CONTAINER_DB);
- containerStore =
- MetadataStoreBuilder.newBuilder()
- .setConf(conf)
- .setDbFile(containerDBPath)
- .setCacheSize(this.cacheSize * OzoneConsts.MB)
- .build();
+ public SCMContainerManager(final Configuration conf,
+ final NodeManager nodeManager, final EventPublisher eventPublisher)
+ throws IOException {
- this.lock = new ReentrantLock();
+ final File metaDir = getOzoneMetaDirPath(conf);
+ final File containerDBPath = new File(metaDir, SCM_CONTAINER_DB);
+ final int cacheSize = conf.getInt(OZONE_SCM_DB_CACHE_SIZE_MB,
+ OZONE_SCM_DB_CACHE_SIZE_DEFAULT);
- size = (long)conf.getStorageSize(OZONE_SCM_CONTAINER_SIZE,
- OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES);
+ this.containerStore = MetadataStoreBuilder.newBuilder()
+ .setConf(conf)
+ .setDbFile(containerDBPath)
+ .setCacheSize(cacheSize * OzoneConsts.MB)
+ .build();
+ this.lock = new ReentrantLock();
+ this.size = (long) conf.getStorageSize(OZONE_SCM_CONTAINER_SIZE,
+ OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.BYTES);
this.pipelineSelector = new PipelineSelector(nodeManager,
- conf, eventPublisher, cacheSizeMB);
-
- this.containerStateManager =
- new ContainerStateManager(conf, this, pipelineSelector);
- LOG.trace("Container State Manager created.");
-
+ conf, eventPublisher, cacheSize);
+ this.containerStateManager = new ContainerStateManager(conf);
this.eventPublisher = eventPublisher;
- long containerCreationLeaseTimeout = conf.getTimeDuration(
+ final long containerCreationLeaseTimeout = conf.getTimeDuration(
ScmConfigKeys.OZONE_SCM_CONTAINER_CREATION_LEASE_TIMEOUT,
ScmConfigKeys.OZONE_SCM_CONTAINER_CREATION_LEASE_TIMEOUT_DEFAULT,
TimeUnit.MILLISECONDS);
- containerLeaseManager = new LeaseManager<>("ContainerCreation",
+ this.containerLeaseManager = new LeaseManager<>("ContainerCreation",
containerCreationLeaseTimeout);
- containerLeaseManager.start();
+ this.containerLeaseManager.start();
+
loadExistingContainers();
}
- private void loadExistingContainers() {
-
- List<ContainerInfo> containerList;
- try {
- containerList = listContainer(0, Integer.MAX_VALUE);
-
- // if there are no container to load, let us return.
- if (containerList == null || containerList.size() == 0) {
- LOG.info("No containers to load for this cluster.");
- return;
- }
- } catch (IOException e) {
- if (!e.getMessage().equals("No container exists in current db")) {
- LOG.error("Could not list the containers", e);
- }
- return;
+ private void loadExistingContainers() throws IOException {
+ List<Map.Entry<byte[], byte[]>> range = containerStore
+ .getSequentialRangeKVs(null, Integer.MAX_VALUE, null);
+ for (Map.Entry<byte[], byte[]> entry : range) {
+ ContainerInfo container = ContainerInfo.fromProtobuf(
+ HddsProtos.SCMContainerInfo.PARSER.parseFrom(entry.getValue()));
+ Preconditions.checkNotNull(container);
+ containerStateManager.loadContainer(container);
+ pipelineSelector.addContainerToPipeline(
+ container.getPipelineID(), container.getContainerID());
}
+ }
+ @VisibleForTesting
+ // TODO: remove this later.
+ public ContainerStateManager getContainerStateManager() {
+ return containerStateManager;
+ }
+
+ @Override
+ public List<ContainerInfo> getContainers() {
+ lock.lock();
try {
- for (ContainerInfo container : containerList) {
- containerStateManager.addExistingContainer(container);
- pipelineSelector.addContainerToPipeline(
- container.getPipelineID(), container.getContainerID());
- }
- } catch (SCMException ex) {
- LOG.error("Unable to create a container information. ", ex);
- // Fix me, what is the proper shutdown procedure for SCM ??
- // System.exit(1) // Should we exit here?
+ return containerStateManager.getAllContainerIDs().stream().map(id -> {
+ try {
+ return containerStateManager.getContainer(id);
+ } catch (ContainerNotFoundException e) {
+ // How can this happen?
+ return null;
+ }
+ }).filter(Objects::nonNull).collect(Collectors.toList());
+ } finally {
+ lock.unlock();
}
}
- /**
- * {@inheritDoc}
- */
@Override
- public ContainerInfo getContainer(final long containerID) throws
- IOException {
- ContainerInfo containerInfo;
+ public List<ContainerInfo> getContainers(LifeCycleState state) {
lock.lock();
try {
- byte[] containerBytes = containerStore.get(
- Longs.toByteArray(containerID));
- if (containerBytes == null) {
- throw new SCMException(
- "Specified key does not exist. key : " + containerID,
- SCMException.ResultCodes.FAILED_TO_FIND_CONTAINER);
- }
-
- HddsProtos.SCMContainerInfo temp = HddsProtos.SCMContainerInfo.PARSER
- .parseFrom(containerBytes);
- containerInfo = ContainerInfo.fromProtobuf(temp);
- return containerInfo;
+ return containerStateManager.getContainerIDsByState(state).stream()
+ .map(id -> {
+ try {
+ return containerStateManager.getContainer(id);
+ } catch (ContainerNotFoundException e) {
+ // How can this happen?
+ return null;
+ }
+ }).filter(Objects::nonNull).collect(Collectors.toList());
} finally {
lock.unlock();
}
}
/**
+ * {@inheritDoc}
+ */
+ @Override
+ public ContainerInfo getContainer(final ContainerID containerID)
+ throws ContainerNotFoundException {
+ return containerStateManager.getContainer(containerID);
+ }
+
+ /**
* Returns the ContainerInfo and pipeline from the containerID. If container
* has no available replicas in datanodes it returns pipeline with no
* datanodes and empty leaderID . Pipeline#isEmpty can be used to check for
@@ -217,38 +213,29 @@ public class SCMContainerManager implements ContainerManager {
* @throws IOException
*/
@Override
- public ContainerWithPipeline getContainerWithPipeline(long containerID)
- throws IOException {
- ContainerInfo contInfo;
+ public ContainerWithPipeline getContainerWithPipeline(ContainerID containerID)
+ throws ContainerNotFoundException {
lock.lock();
try {
- byte[] containerBytes = containerStore.get(
- Longs.toByteArray(containerID));
- if (containerBytes == null) {
- throw new SCMException(
- "Specified key does not exist. key : " + containerID,
- SCMException.ResultCodes.FAILED_TO_FIND_CONTAINER);
- }
- HddsProtos.SCMContainerInfo temp = HddsProtos.SCMContainerInfo.PARSER
- .parseFrom(containerBytes);
- contInfo = ContainerInfo.fromProtobuf(temp);
-
+ final ContainerInfo contInfo = getContainer(containerID);
Pipeline pipeline;
String leaderId = "";
- if (contInfo.isContainerOpen()) {
+ if (contInfo.isOpen()) {
// If pipeline with given pipeline Id already exist return it
pipeline = pipelineSelector.getPipeline(contInfo.getPipelineID());
} else {
// For close containers create pipeline from datanodes with replicas
- Set<DatanodeDetails> dnWithReplicas = containerStateManager
+ Set<ContainerReplica> dnWithReplicas = containerStateManager
.getContainerReplicas(contInfo.containerID());
if (!dnWithReplicas.isEmpty()) {
- leaderId = dnWithReplicas.iterator().next().getUuidString();
+ leaderId = dnWithReplicas.iterator().next()
+ .getDatanodeDetails().getUuidString();
}
pipeline = new Pipeline(leaderId, contInfo.getState(),
ReplicationType.STAND_ALONE, contInfo.getReplicationFactor(),
PipelineID.randomId());
- dnWithReplicas.forEach(pipeline::addMember);
+ dnWithReplicas.stream().map(ContainerReplica::getDatanodeDetails).
+ forEach(pipeline::addMember);
}
return new ContainerWithPipeline(contInfo, pipeline);
} finally {
@@ -260,33 +247,32 @@ public class SCMContainerManager implements ContainerManager {
* {@inheritDoc}
*/
@Override
- public List<ContainerInfo> listContainer(long startContainerID,
- int count) throws IOException {
- List<ContainerInfo> containerList = new ArrayList<>();
+ public List<ContainerInfo> listContainer(ContainerID startContainerID,
+ int count) {
lock.lock();
try {
- if (containerStore.isEmpty()) {
- throw new IOException("No container exists in current db");
- }
- byte[] startKey = startContainerID <= 0 ? null :
- Longs.toByteArray(startContainerID);
- List<Map.Entry<byte[], byte[]>> range =
- containerStore.getSequentialRangeKVs(startKey, count, null);
-
- // Transform the values into the pipelines.
- // TODO: filter by container state
- for (Map.Entry<byte[], byte[]> entry : range) {
- ContainerInfo containerInfo =
- ContainerInfo.fromProtobuf(
- HddsProtos.SCMContainerInfo.PARSER.parseFrom(
- entry.getValue()));
- Preconditions.checkNotNull(containerInfo);
- containerList.add(containerInfo);
- }
+ final long startId = startContainerID == null ?
+ 0 : startContainerID.getId();
+ final List<ContainerID> containersIds =
+ new ArrayList<>(containerStateManager.getAllContainerIDs());
+ Collections.sort(containersIds);
+
+ return containersIds.stream()
+ .filter(id -> id.getId() > startId)
+ .limit(count)
+ .map(id -> {
+ try {
+ return containerStateManager.getContainer(id);
+ } catch (ContainerNotFoundException ex) {
+ // This can never happen, as we hold lock no one else can remove
+ // the container after we got the container ids.
+ LOG.warn("Container Missing.", ex);
+ return null;
+ }
+ }).collect(Collectors.toList());
} finally {
lock.unlock();
}
- return containerList;
}
/**
@@ -298,29 +284,35 @@ public class SCMContainerManager implements ContainerManager {
* @throws IOException - Exception
*/
@Override
- public ContainerWithPipeline allocateContainer(
- ReplicationType type,
- ReplicationFactor replicationFactor,
- String owner)
+ public ContainerWithPipeline allocateContainer(final ReplicationType type,
+ final ReplicationFactor replicationFactor, final String owner)
throws IOException {
-
- ContainerInfo containerInfo;
- ContainerWithPipeline containerWithPipeline;
-
lock.lock();
try {
- containerWithPipeline = containerStateManager.allocateContainer(
- pipelineSelector, type, replicationFactor, owner);
- containerInfo = containerWithPipeline.getContainerInfo();
-
- byte[] containerIDBytes = Longs.toByteArray(
- containerInfo.getContainerID());
- containerStore.put(containerIDBytes, containerInfo.getProtobuf()
- .toByteArray());
+ final ContainerInfo containerInfo; containerInfo = containerStateManager
+ .allocateContainer(pipelineSelector, type, replicationFactor, owner);
+ final Pipeline pipeline = pipelineSelector.getPipeline(
+ containerInfo.getPipelineID());
+
+ try {
+ final byte[] containerIDBytes = Longs.toByteArray(
+ containerInfo.getContainerID());
+ containerStore.put(containerIDBytes,
+ containerInfo.getProtobuf().toByteArray());
+ } catch (IOException ex) {
+ // If adding to containerStore fails, we should remove the container
+ // from in-memory map.
+ try {
+ containerStateManager.removeContainer(containerInfo.containerID());
+ } catch (ContainerNotFoundException cnfe) {
+ // No need to worry much, everything is going as planned.
+ }
+ throw ex;
+ }
+ return new ContainerWithPipeline(containerInfo, pipeline);
} finally {
lock.unlock();
}
- return containerWithPipeline;
}
/**
@@ -332,18 +324,24 @@ public class SCMContainerManager implements ContainerManager {
* specified key.
*/
@Override
- public void deleteContainer(long containerID) throws IOException {
+ public void deleteContainer(ContainerID containerID) throws IOException {
lock.lock();
try {
- byte[] dbKey = Longs.toByteArray(containerID);
- byte[] containerBytes = containerStore.get(dbKey);
- if (containerBytes == null) {
- throw new SCMException(
- "Failed to delete container " + containerID + ", reason : " +
- "container doesn't exist.",
- SCMException.ResultCodes.FAILED_TO_FIND_CONTAINER);
+ containerStateManager.removeContainer(containerID);
+ final byte[] dbKey = Longs.toByteArray(containerID.getId());
+ final byte[] containerBytes = containerStore.get(dbKey);
+ if (containerBytes != null) {
+ containerStore.delete(dbKey);
+ } else {
+ // Where did the container go? o_O
+ LOG.warn("Unable to remove the container {} from container store," +
+ " it's missing!", containerID);
}
- containerStore.delete(dbKey);
+ } catch (ContainerNotFoundException cnfe) {
+ throw new SCMException(
+ "Failed to delete container " + containerID + ", reason : " +
+ "container doesn't exist.",
+ SCMException.ResultCodes.FAILED_TO_FIND_CONTAINER);
} finally {
lock.unlock();
}
@@ -354,40 +352,51 @@ public class SCMContainerManager implements ContainerManager {
*/
@Override
public HddsProtos.LifeCycleState updateContainerState(
- long containerID, HddsProtos.LifeCycleEvent event) throws
- IOException {
- ContainerInfo containerInfo;
+ ContainerID containerID, HddsProtos.LifeCycleEvent event)
+ throws IOException {
+ // Should we return the updated ContainerInfo instead of LifeCycleState?
lock.lock();
try {
- byte[] dbKey = Longs.toByteArray(containerID);
- byte[] containerBytes = containerStore.get(dbKey);
- if (containerBytes == null) {
- throw new SCMException(
- "Failed to update container state"
- + containerID
- + ", reason : container doesn't exist.",
- SCMException.ResultCodes.FAILED_TO_FIND_CONTAINER);
+ ContainerInfo updatedContainer =
+ updateContainerStateInternal(containerID, event);
+ if (!updatedContainer.isOpen()) {
+ pipelineSelector.removeContainerFromPipeline(
+ updatedContainer.getPipelineID(), containerID.getId());
}
- containerInfo =
- ContainerInfo.fromProtobuf(HddsProtos.SCMContainerInfo.PARSER
- .parseFrom(containerBytes));
+ final byte[] dbKey = Longs.toByteArray(containerID.getId());
+ containerStore.put(dbKey, updatedContainer.getProtobuf().toByteArray());
+ return updatedContainer.getState();
+ } catch (ContainerNotFoundException cnfe) {
+ throw new SCMException(
+ "Failed to update container state"
+ + containerID
+ + ", reason : container doesn't exist.",
+ SCMException.ResultCodes.FAILED_TO_FIND_CONTAINER);
+ } finally {
+ lock.unlock();
+ }
+ }
- Preconditions.checkNotNull(containerInfo);
+ private ContainerInfo updateContainerStateInternal(ContainerID containerID,
+ HddsProtos.LifeCycleEvent event) throws IOException {
+ // Refactor the below code for better clarity.
+ try {
+ final ContainerInfo info =
+ containerStateManager.getContainer(containerID);
switch (event) {
case CREATE:
// Acquire lease on container
Lease<ContainerInfo> containerLease =
- containerLeaseManager.acquire(containerInfo);
+ containerLeaseManager.acquire(info);
// Register callback to be executed in case of timeout
containerLease.registerCallBack(() -> {
updateContainerState(containerID,
HddsProtos.LifeCycleEvent.TIMEOUT);
- return null;
- });
+ return null; });
break;
case CREATED:
// Release the lease on container
- containerLeaseManager.release(containerInfo);
+ containerLeaseManager.release(info);
break;
case FINALIZE:
// TODO: we don't need a lease manager here for closing as the
@@ -412,28 +421,20 @@ public class SCMContainerManager implements ContainerManager {
// If the below updateContainerState call fails, we should revert the
// changes made in switch case.
// Like releasing the lease in case of BEGIN_CREATE.
- ContainerInfo updatedContainer = containerStateManager
- .updateContainerState(containerInfo, event);
- if (!updatedContainer.isContainerOpen()) {
- pipelineSelector.removeContainerFromPipeline(
- containerInfo.getPipelineID(), containerID);
- }
- containerStore.put(dbKey, updatedContainer.getProtobuf().toByteArray());
- return updatedContainer.getState();
+ return containerStateManager.updateContainerState(containerID, event);
} catch (LeaseException e) {
throw new IOException("Lease Exception.", e);
- } finally {
- lock.unlock();
}
}
- /**
- * Update deleteTransactionId according to deleteTransactionMap.
- *
- * @param deleteTransactionMap Maps the containerId to latest delete
- * transaction id for the container.
- * @throws IOException
- */
+
+ /**
+ * Update deleteTransactionId according to deleteTransactionMap.
+ *
+ * @param deleteTransactionMap Maps the containerId to latest delete
+ * transaction id for the container.
+ * @throws IOException
+ */
public void updateDeleteTransactionId(Map<Long, Long> deleteTransactionMap)
throws IOException {
if (deleteTransactionMap == null) {
@@ -467,16 +468,6 @@ public class SCMContainerManager implements ContainerManager {
}
/**
- * Returns the container State Manager.
- *
- * @return ContainerStateManager
- */
- @Override
- public ContainerStateManager getStateManager() {
- return containerStateManager;
- }
-
- /**
* Return a container matching the attributes specified.
*
* @param sizeRequired - Space needed in the Container.
@@ -489,7 +480,7 @@ public class SCMContainerManager implements ContainerManager {
public ContainerWithPipeline getMatchingContainerWithPipeline(
final long sizeRequired, String owner, ReplicationType type,
ReplicationFactor factor, LifeCycleState state) throws IOException {
- ContainerInfo containerInfo = getStateManager()
+ ContainerInfo containerInfo = containerStateManager
.getMatchingContainer(sizeRequired, owner, type, factor, state);
if (containerInfo == null) {
return null;
@@ -518,70 +509,45 @@ public class SCMContainerManager implements ContainerManager {
*/
@Override
public void processContainerReports(DatanodeDetails datanodeDetails,
- ContainerReportsProto reports, boolean isRegisterCall)
- throws IOException {
+ ContainerReportsProto reports) throws IOException {
List<StorageContainerDatanodeProtocolProtos.ContainerInfo>
containerInfos = reports.getReportsList();
PendingDeleteStatusList pendingDeleteStatusList =
new PendingDeleteStatusList(datanodeDetails);
- for (StorageContainerDatanodeProtocolProtos.ContainerInfo contInfo :
+ for (StorageContainerDatanodeProtocolProtos.ContainerInfo newInfo :
containerInfos) {
- // Update replica info during registration process.
- if (isRegisterCall) {
- try {
- getStateManager().addContainerReplica(ContainerID.
- valueof(contInfo.getContainerID()), datanodeDetails);
- } catch (Exception ex) {
- // Continue to next one after logging the error.
- LOG.error("Error while adding replica for containerId {}.",
- contInfo.getContainerID(), ex);
- }
- }
- byte[] dbKey = Longs.toByteArray(contInfo.getContainerID());
+ ContainerID id = ContainerID.valueof(newInfo.getContainerID());
+ ContainerReplica replica = ContainerReplica.newBuilder()
+ .setContainerID(id)
+ .setDatanodeDetails(datanodeDetails)
+ .setOriginNodeId(datanodeDetails.getUuid())
+ .build();
lock.lock();
try {
- byte[] containerBytes = containerStore.get(dbKey);
- if (containerBytes != null) {
- HddsProtos.SCMContainerInfo knownState =
- HddsProtos.SCMContainerInfo.PARSER.parseFrom(containerBytes);
-
- if (knownState.getState() == LifeCycleState.CLOSING
- && contInfo.getState() == LifeCycleState.CLOSED) {
-
- updateContainerState(contInfo.getContainerID(),
- LifeCycleEvent.CLOSE);
-
- //reread the container
- knownState =
- HddsProtos.SCMContainerInfo.PARSER
- .parseFrom(containerStore.get(dbKey));
- }
-
- HddsProtos.SCMContainerInfo newState =
- reconcileState(contInfo, knownState, datanodeDetails);
-
- if (knownState.getDeleteTransactionId() > contInfo
- .getDeleteTransactionId()) {
- pendingDeleteStatusList
- .addPendingDeleteStatus(contInfo.getDeleteTransactionId(),
- knownState.getDeleteTransactionId(),
- knownState.getContainerID());
- }
-
- // FIX ME: This can be optimized, we write twice to memory, where a
- // single write would work well.
- //
- // We need to write this to DB again since the closed only write
- // the updated State.
- containerStore.put(dbKey, newState.toByteArray());
-
- } else {
- // Container not found in our container db.
- LOG.error("Error while processing container report from datanode :" +
- " {}, for container: {}, reason: container doesn't exist in" +
- "container database.", datanodeDetails,
- contInfo.getContainerID());
+ containerStateManager.updateContainerReplica(id, replica);
+ ContainerInfo currentInfo = containerStateManager.getContainer(id);
+ if (newInfo.getState() == LifeCycleState.CLOSING
+ && currentInfo.getState() == LifeCycleState.CLOSED) {
+ currentInfo = updateContainerStateInternal(id, LifeCycleEvent.CLOSE);
}
+
+ HddsProtos.SCMContainerInfo newState =
+ reconcileState(newInfo, currentInfo);
+
+ if (currentInfo.getDeleteTransactionId() >
+ newInfo.getDeleteTransactionId()) {
+ pendingDeleteStatusList
+ .addPendingDeleteStatus(newInfo.getDeleteTransactionId(),
+ currentInfo.getDeleteTransactionId(),
+ currentInfo.getContainerID());
+ }
+ containerStateManager.updateContainerInfo(
+ ContainerInfo.fromProtobuf(newState));
+ containerStore.put(id.getBytes(), newState.toByteArray());
+ } catch (ContainerNotFoundException e) {
+ LOG.error("Error while processing container report from datanode :" +
+ " {}, for container: {}, reason: container doesn't exist in" +
+ "container database.", datanodeDetails, id);
} finally {
lock.unlock();
}
@@ -598,36 +564,21 @@ public class SCMContainerManager implements ContainerManager {
*
* @param datanodeState - State from the Datanode.
* @param knownState - State inside SCM.
- * @param dnDetails
* @return new SCM State for this container.
*/
private HddsProtos.SCMContainerInfo reconcileState(
StorageContainerDatanodeProtocolProtos.ContainerInfo datanodeState,
- SCMContainerInfo knownState, DatanodeDetails dnDetails) {
+ ContainerInfo knownState) {
HddsProtos.SCMContainerInfo.Builder builder =
HddsProtos.SCMContainerInfo.newBuilder();
builder.setContainerID(knownState.getContainerID())
- .setPipelineID(knownState.getPipelineID())
+ .setPipelineID(knownState.getPipelineID().getProtobuf())
+ .setState(knownState.getState())
.setReplicationType(knownState.getReplicationType())
- .setReplicationFactor(knownState.getReplicationFactor());
-
- // TODO: If current state doesn't have this DN in list of DataNodes with
- // replica then add it in list of replicas.
-
- // If used size is greater than allocated size, we will be updating
- // allocated size with used size. This update is done as a fallback
- // mechanism in case SCM crashes without properly updating allocated
- // size. Correct allocated value will be updated by
- // ContainerStateManager during SCM shutdown.
- long usedSize = datanodeState.getUsed();
- long allocated = knownState.getAllocatedBytes() > usedSize ?
- knownState.getAllocatedBytes() : usedSize;
- builder.setAllocatedBytes(allocated)
- .setUsedBytes(usedSize)
+ .setReplicationFactor(knownState.getReplicationFactor())
+ .setUsedBytes(datanodeState.getUsed())
.setNumberOfKeys(datanodeState.getKeyCount())
- .setState(knownState.getState())
.setStateEnterTime(knownState.getStateEnterTime())
- .setContainerID(knownState.getContainerID())
.setDeleteTransactionId(knownState.getDeleteTransactionId());
if (knownState.getOwner() != null) {
builder.setOwner(knownState.getOwner());
@@ -635,20 +586,40 @@ public class SCMContainerManager implements ContainerManager {
return builder.build();
}
+ /**
+ * Returns the latest list of DataNodes where replica for given containerId
+ * exist. Throws an SCMException if no entry is found for given containerId.
+ *
+ * @param containerID
+ * @return Set<DatanodeDetails>
+ */
+ public Set<ContainerReplica> getContainerReplicas(
+ final ContainerID containerID) throws ContainerNotFoundException {
+ return containerStateManager.getContainerReplicas(containerID);
+ }
/**
- * In Container is in closed state, if it is in closed, Deleting or Deleted
- * State.
+ * Add a container Replica for given DataNode.
*
- * @param info - ContainerInfo.
- * @return true if is in open state, false otherwise
+ * @param containerID
+ * @param replica
*/
- private boolean shouldClose(ContainerInfo info) {
- return info.getState() == HddsProtos.LifeCycleState.OPEN;
+ public void updateContainerReplica(final ContainerID containerID,
+ final ContainerReplica replica) throws ContainerNotFoundException {
+ containerStateManager.updateContainerReplica(containerID, replica);
}
- private boolean isClosed(ContainerInfo info) {
- return info.getState() == HddsProtos.LifeCycleState.CLOSED;
+ /**
+ * Remove a container Replica for given DataNode.
+ *
+ * @param containerID
+ * @param replica
+ * @return True of dataNode is removed successfully else false.
+ */
+ public void removeContainerReplica(final ContainerID containerID,
+ final ContainerReplica replica)
+ throws ContainerNotFoundException, ContainerReplicaNotFoundException {
+ containerStateManager.removeContainerReplica(containerID, replica);
}
/**
@@ -671,7 +642,6 @@ public class SCMContainerManager implements ContainerManager {
containerLeaseManager.shutdown();
}
if (containerStateManager != null) {
- flushContainerInfo();
containerStateManager.close();
}
if (containerStore != null) {
@@ -683,48 +653,6 @@ public class SCMContainerManager implements ContainerManager {
}
}
- /**
- * Since allocatedBytes of a container is only in memory, stored in
- * containerStateManager, when closing SCMContainerManager, we need to update
- * this in the container store.
- *
- * @throws IOException on failure.
- */
- @VisibleForTesting
- public void flushContainerInfo() throws IOException {
- List<ContainerInfo> containers = containerStateManager.getAllContainers();
- List<Long> failedContainers = new ArrayList<>();
- for (ContainerInfo info : containers) {
- // even if some container updated failed, others can still proceed
- try {
- byte[] dbKey = Longs.toByteArray(info.getContainerID());
- byte[] containerBytes = containerStore.get(dbKey);
- // TODO : looks like when a container is deleted, the container is
- // removed from containerStore but not containerStateManager, so it can
- // return info of a deleted container. may revisit this in the future,
- // for now, just skip a not-found container
- if (containerBytes != null) {
- containerStore.put(dbKey, info.getProtobuf().toByteArray());
- } else {
- LOG.debug("Container state manager has container {} but not found " +
- "in container store, a deleted container?",
- info.getContainerID());
- }
- } catch (IOException ioe) {
- failedContainers.add(info.getContainerID());
- }
- }
- if (!failedContainers.isEmpty()) {
- throw new IOException("Error in flushing container info from container " +
- "state manager: " + failedContainers);
- }
- }
-
- @VisibleForTesting
- public MetadataStore getContainerStore() {
- return containerStore;
- }
-
public PipelineSelector getPipelineSelector() {
return pipelineSelector;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/50715c06/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java
index ddecdbc..8c11e84 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java
@@ -22,12 +22,14 @@ import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ThreadFactory;
+import java.util.stream.Collectors;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
import org.apache.hadoop.hdds.scm.container.ContainerID;
-import org.apache.hadoop.hdds.scm.container.ContainerStateManager;
-import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerManager;
+import org.apache.hadoop.hdds.scm.container.ContainerReplica;
import org.apache.hadoop.hdds.scm.container.placement.algorithms
.ContainerPlacementPolicy;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
@@ -64,14 +66,14 @@ public class ReplicationManager implements Runnable {
private boolean running = true;
- private ContainerStateManager containerStateManager;
+ private ContainerManager containerManager;
public ReplicationManager(ContainerPlacementPolicy containerPlacement,
- ContainerStateManager containerStateManager, EventQueue eventQueue,
+ ContainerManager containerManager, EventQueue eventQueue,
LeaseManager<Long> commandWatcherLeaseManager) {
this.containerPlacement = containerPlacement;
- this.containerStateManager = containerStateManager;
+ this.containerManager = containerManager;
this.eventPublisher = eventQueue;
this.replicationCommandWatcher =
@@ -106,7 +108,7 @@ public class ReplicationManager implements Runnable {
ContainerID containerID = new ContainerID(request.getContainerId());
ContainerInfo containerInfo =
- containerStateManager.getContainer(containerID);
+ containerManager.getContainer(containerID);
Preconditions.checkNotNull(containerInfo,
"No information about the container " + request.getContainerId());
@@ -116,10 +118,10 @@ public class ReplicationManager implements Runnable {
"Container should be in closed state");
//check the current replication
- List<DatanodeDetails> datanodesWithReplicas =
+ List<ContainerReplica> containerReplicas =
new ArrayList<>(getCurrentReplicas(request));
- if (datanodesWithReplicas.size() == 0) {
+ if (containerReplicas.size() == 0) {
LOG.warn(
"Container {} should be replicated but can't find any existing "
+ "replicas",
@@ -134,21 +136,23 @@ public class ReplicationManager implements Runnable {
.size();
int deficit =
- request.getExpecReplicationCount() - datanodesWithReplicas.size()
+ request.getExpecReplicationCount() - containerReplicas.size()
- inFlightReplications;
if (deficit > 0) {
+ List<DatanodeDetails> datanodes = containerReplicas.stream()
+ .map(ContainerReplica::getDatanodeDetails)
+ .collect(Collectors.toList());
List<DatanodeDetails> selectedDatanodes = containerPlacement
- .chooseDatanodes(datanodesWithReplicas, deficit,
+ .chooseDatanodes(datanodes, deficit,
containerInfo.getUsedBytes());
//send the command
for (DatanodeDetails datanode : selectedDatanodes) {
ReplicateContainerCommand replicateCommand =
- new ReplicateContainerCommand(containerID.getId(),
- datanodesWithReplicas);
+ new ReplicateContainerCommand(containerID.getId(), datanodes);
eventPublisher.fireEvent(SCMEvents.DATANODE_COMMAND,
new CommandForDatanode<>(
@@ -174,9 +178,9 @@ public class ReplicationManager implements Runnable {
}
@VisibleForTesting
- protected Set<DatanodeDetails> getCurrentReplicas(ReplicationRequest request)
+ protected Set<ContainerReplica> getCurrentReplicas(ReplicationRequest request)
throws IOException {
- return containerStateManager
+ return containerManager
.getContainerReplicas(new ContainerID(request.getContainerId()));
}
@@ -234,7 +238,11 @@ public class ReplicationManager implements Runnable {
}
}
- public static class ReplicationCompleted implements IdentifiableEventPayload {
+ /**
+ * Add javadoc.
+ */
+ public static class ReplicationCompleted
+ implements IdentifiableEventPayload {
private final long uuid;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/50715c06/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java
index 880a715..b8052a4 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java
@@ -20,19 +20,21 @@ package org.apache.hadoop.hdds.scm.container.states;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import java.util.HashSet;
-import java.util.Set;
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+
import org.apache.hadoop.hdds.scm.container.ContainerID;
-import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException;
+import org.apache.hadoop.hdds.scm.container.ContainerReplica;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerReplicaNotFoundException;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
-import org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.HashSet;
+import java.util.Set;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -46,8 +48,6 @@ import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes
.CONTAINER_EXISTS;
import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes
.FAILED_TO_CHANGE_CONTAINER_STATE;
-import static org.apache.hadoop.hdds.scm.exceptions.SCMException.ResultCodes
- .FAILED_TO_FIND_CONTAINER;
/**
* Container State Map acts like a unified map for various attributes that are
@@ -84,16 +84,15 @@ public class ContainerStateMap {
private static final Logger LOG =
LoggerFactory.getLogger(ContainerStateMap.class);
+ private final static NavigableSet<ContainerID> EMPTY_SET =
+ Collections.unmodifiableNavigableSet(new TreeSet<>());
+
private final ContainerAttribute<LifeCycleState> lifeCycleStateMap;
private final ContainerAttribute<String> ownerMap;
private final ContainerAttribute<ReplicationFactor> factorMap;
private final ContainerAttribute<ReplicationType> typeMap;
-
private final Map<ContainerID, ContainerInfo> containerMap;
- // Map to hold replicas of given container.
- private final Map<ContainerID, Set<DatanodeDetails>> contReplicaMap;
- private final static NavigableSet<ContainerID> EMPTY_SET =
- Collections.unmodifiableNavigableSet(new TreeSet<>());
+ private final Map<ContainerID, Set<ContainerReplica>> replicaMap;
private final Map<ContainerQueryKey, NavigableSet<ContainerID>> resultCache;
// Container State Map lock should be held before calling into
@@ -105,18 +104,14 @@ public class ContainerStateMap {
* Create a ContainerStateMap.
*/
public ContainerStateMap() {
- lifeCycleStateMap = new ContainerAttribute<>();
- ownerMap = new ContainerAttribute<>();
- factorMap = new ContainerAttribute<>();
- typeMap = new ContainerAttribute<>();
- containerMap = new HashMap<>();
- lock = new ReentrantReadWriteLock();
- contReplicaMap = new HashMap<>();
-// new InstrumentedLock(getClass().getName(), LOG,
-// new ReentrantLock(),
-// 1000,
-// 300));
- resultCache = new ConcurrentHashMap<>();
+ this.lifeCycleStateMap = new ContainerAttribute<>();
+ this.ownerMap = new ContainerAttribute<>();
+ this.factorMap = new ContainerAttribute<>();
+ this.typeMap = new ContainerAttribute<>();
+ this.containerMap = new HashMap<>();
+ this.lock = new ReentrantReadWriteLock();
+ this.replicaMap = new HashMap<>();
+ this.resultCache = new ConcurrentHashMap<>();
}
/**
@@ -125,7 +120,7 @@ public class ContainerStateMap {
* @param info - container info
* @throws SCMException - throws if create failed.
*/
- public void addContainer(ContainerInfo info)
+ public void addContainer(final ContainerInfo info)
throws SCMException {
Preconditions.checkNotNull(info, "Container Info cannot be null");
Preconditions.checkArgument(info.getReplicationFactor().getNumber() > 0,
@@ -133,7 +128,7 @@ public class ContainerStateMap {
lock.writeLock().lock();
try {
- ContainerID id = ContainerID.valueof(info.getContainerID());
+ final ContainerID id = info.containerID();
if (containerMap.putIfAbsent(id, info) != null) {
LOG.debug("Duplicate container ID detected. {}", id);
throw new
@@ -145,6 +140,7 @@ public class ContainerStateMap {
ownerMap.insert(info.getOwner(), id);
factorMap.insert(info.getReplicationFactor(), id);
typeMap.insert(info.getReplicationType(), id);
+ replicaMap.put(id, new HashSet<>());
// Flush the cache of this container type, will be added later when
// get container queries are executed.
@@ -156,23 +152,30 @@ public class ContainerStateMap {
}
/**
- * Returns the latest state of Container from SCM's Container State Map.
+ * Removes a Container Entry from ContainerStateMap.
*
- * @param info - ContainerInfo
- * @return ContainerInfo
- */
- public ContainerInfo getContainerInfo(ContainerInfo info) {
- return getContainerInfo(info.getContainerID());
- }
-
- /**
- * Returns the latest state of Container from SCM's Container State Map.
- *
- * @param containerID - int
- * @return container info, if found.
+ * @param containerID - ContainerID
+ * @throws SCMException - throws if create failed.
*/
- public ContainerInfo getContainerInfo(long containerID) {
- return getContainerInfo(ContainerID.valueof(containerID));
+ public void removeContainer(final ContainerID containerID)
+ throws ContainerNotFoundException {
+ Preconditions.checkNotNull(containerID, "ContainerID cannot be null");
+ lock.writeLock().lock();
+ try {
+ checkIfContainerExist(containerID);
+ // Should we revert back to the original state if any of the below
+ // remove operation fails?
+ final ContainerInfo info = containerMap.remove(containerID);
+ lifeCycleStateMap.remove(info.getState(), containerID);
+ ownerMap.remove(info.getOwner(), containerID);
+ factorMap.remove(info.getReplicationFactor(), containerID);
+ typeMap.remove(info.getReplicationType(), containerID);
+ // Flush the cache of this container type.
+ flushCache(info);
+ LOG.trace("Removed container with {} successfully.", containerID);
+ } finally {
+ lock.writeLock().unlock();
+ }
}
/**
@@ -181,9 +184,11 @@ public class ContainerStateMap {
* @param containerID - ContainerID
* @return container info, if found.
*/
- public ContainerInfo getContainerInfo(ContainerID containerID) {
+ public ContainerInfo getContainerInfo(final ContainerID containerID)
+ throws ContainerNotFoundException {
lock.readLock().lock();
try {
+ checkIfContainerExist(containerID);
return containerMap.get(containerID);
} finally {
lock.readLock().unlock();
@@ -197,21 +202,17 @@ public class ContainerStateMap {
* @param containerID
* @return Set<DatanodeDetails>
*/
- public Set<DatanodeDetails> getContainerReplicas(ContainerID containerID)
- throws SCMException {
+ public Set<ContainerReplica> getContainerReplicas(
+ final ContainerID containerID) throws ContainerNotFoundException {
Preconditions.checkNotNull(containerID);
lock.readLock().lock();
try {
- if (contReplicaMap.containsKey(containerID)) {
- return Collections
- .unmodifiableSet(contReplicaMap.get(containerID));
- }
+ checkIfContainerExist(containerID);
+ return Collections
+ .unmodifiableSet(new HashSet<>(replicaMap.get(containerID)));
} finally {
lock.readLock().unlock();
}
- throw new SCMException(
- "No entry exist for containerId: " + containerID + " in replica map.",
- ResultCodes.NO_REPLICA_FOUND);
}
/**
@@ -220,26 +221,17 @@ public class ContainerStateMap {
* ContainerId.
*
* @param containerID
- * @param dnList
+ * @param replica
*/
- public void addContainerReplica(ContainerID containerID,
- DatanodeDetails... dnList) {
+ public void updateContainerReplica(final ContainerID containerID,
+ final ContainerReplica replica) throws ContainerNotFoundException {
Preconditions.checkNotNull(containerID);
lock.writeLock().lock();
try {
- for (DatanodeDetails dn : dnList) {
- Preconditions.checkNotNull(dn);
- if (contReplicaMap.containsKey(containerID)) {
- if(!contReplicaMap.get(containerID).add(dn)) {
- LOG.debug("ReplicaMap already contains entry for container Id: "
- + "{},DataNode: {}", containerID, dn);
- }
- } else {
- Set<DatanodeDetails> dnSet = new HashSet<>();
- dnSet.add(dn);
- contReplicaMap.put(containerID, dnSet);
- }
- }
+ checkIfContainerExist(containerID);
+ Set<ContainerReplica> replicas = replicaMap.get(containerID);
+ replicas.remove(replica);
+ replicas.add(replica);
} finally {
lock.writeLock().unlock();
}
@@ -249,61 +241,45 @@ public class ContainerStateMap {
* Remove a container Replica for given DataNode.
*
* @param containerID
- * @param dn
+ * @param replica
* @return True of dataNode is removed successfully else false.
*/
- public boolean removeContainerReplica(ContainerID containerID,
- DatanodeDetails dn) throws SCMException {
+ public void removeContainerReplica(final ContainerID containerID,
+ final ContainerReplica replica)
+ throws ContainerNotFoundException, ContainerReplicaNotFoundException {
Preconditions.checkNotNull(containerID);
- Preconditions.checkNotNull(dn);
+ Preconditions.checkNotNull(replica);
lock.writeLock().lock();
try {
- if (contReplicaMap.containsKey(containerID)) {
- return contReplicaMap.get(containerID).remove(dn);
+ checkIfContainerExist(containerID);
+ if(!replicaMap.get(containerID).remove(replica)) {
+ throw new ContainerReplicaNotFoundException(
+ "Container #"
+ + containerID.getId() + ", replica: " + replica);
}
} finally {
lock.writeLock().unlock();
}
- throw new SCMException(
- "No entry exist for containerId: " + containerID + " in replica map.",
- ResultCodes.FAILED_TO_FIND_CONTAINER);
}
@VisibleForTesting
+ // TODO: fix the test case and remove this method!
public static Logger getLOG() {
return LOG;
}
/**
- * Returns the full container Map.
- *
- * @return - Map
- */
- public Map<ContainerID, ContainerInfo> getContainerMap() {
- lock.readLock().lock();
- try {
- return Collections.unmodifiableMap(containerMap);
- } finally {
- lock.readLock().unlock();
- }
- }
-
- /**
* Just update the container State.
* @param info ContainerInfo.
*/
- public void updateContainerInfo(ContainerInfo info) throws SCMException {
- Preconditions.checkNotNull(info);
- ContainerInfo currentInfo = null;
+ public void updateContainerInfo(final ContainerInfo info)
+ throws ContainerNotFoundException {
lock.writeLock().lock();
try {
- currentInfo = containerMap.get(
- ContainerID.valueof(info.getContainerID()));
-
- if (currentInfo == null) {
- throw new SCMException("No such container.", FAILED_TO_FIND_CONTAINER);
- }
+ Preconditions.checkNotNull(info);
+ checkIfContainerExist(info.containerID());
+ final ContainerInfo currentInfo = containerMap.get(info.containerID());
flushCache(info, currentInfo);
containerMap.put(info.containerID(), info);
} finally {
@@ -314,33 +290,23 @@ public class ContainerStateMap {
/**
* Update the State of a container.
*
- * @param info - ContainerInfo
+ * @param containerID - ContainerID
* @param currentState - CurrentState
* @param newState - NewState.
* @throws SCMException - in case of failure.
*/
- public void updateState(ContainerInfo info, LifeCycleState currentState,
- LifeCycleState newState) throws SCMException {
+ public void updateState(ContainerID containerID, LifeCycleState currentState,
+ LifeCycleState newState) throws SCMException, ContainerNotFoundException {
Preconditions.checkNotNull(currentState);
Preconditions.checkNotNull(newState);
-
- ContainerID id = new ContainerID(info.getContainerID());
- ContainerInfo currentInfo = null;
-
lock.writeLock().lock();
try {
+ checkIfContainerExist(containerID);
+ final ContainerInfo currentInfo = containerMap.get(containerID);
try {
- // Just flush both old and new data sets from the result cache.
- ContainerInfo newInfo = new ContainerInfo(info);
+ final ContainerInfo newInfo = new ContainerInfo(currentInfo);
newInfo.setState(newState);
- flushCache(newInfo, info);
-
- currentInfo = containerMap.get(id);
- if (currentInfo == null) {
- throw new
- SCMException("No such container.", FAILED_TO_FIND_CONTAINER);
- }
// We are updating two places before this update is done, these can
// fail independently, since the code needs to handle it.
@@ -351,11 +317,13 @@ public class ContainerStateMap {
// roll back the earlier change we did. If the rollback fails, we can
// be in an inconsistent state,
- info.setState(newState);
- containerMap.put(id, info);
- lifeCycleStateMap.update(currentState, newState, id);
+ containerMap.put(containerID, newInfo);
+ lifeCycleStateMap.update(currentState, newState, containerID);
LOG.trace("Updated the container {} to new state. Old = {}, new = " +
- "{}", id, currentState, newState);
+ "{}", containerID, currentState, newState);
+
+ // Just flush both old and new data sets from the result cache.
+ flushCache(currentInfo, newInfo);
} catch (SCMException ex) {
LOG.error("Unable to update the container state. {}", ex);
// we need to revert the change in this attribute since we are not
@@ -364,13 +332,13 @@ public class ContainerStateMap {
"old state. Old = {}, Attempted state = {}", currentState,
newState);
- containerMap.put(id, currentInfo);
+ containerMap.put(containerID, currentInfo);
// if this line throws, the state map can be in an inconsistent
// state, since we will have modified the attribute by the
// container state will not in sync since we were not able to put
// that into the hash table.
- lifeCycleStateMap.update(newState, currentState, id);
+ lifeCycleStateMap.update(newState, currentState, containerID);
throw new SCMException("Updating the container map failed.", ex,
FAILED_TO_CHANGE_CONTAINER_STATE);
@@ -380,13 +348,17 @@ public class ContainerStateMap {
}
}
+ public Set<ContainerID> getAllContainerIDs() {
+ return containerMap.keySet();
+ }
+
/**
* Returns A list of containers owned by a name service.
*
* @param ownerName - Name of the NameService.
* @return - NavigableSet of ContainerIDs.
*/
- NavigableSet<ContainerID> getContainerIDsByOwner(String ownerName) {
+ NavigableSet<ContainerID> getContainerIDsByOwner(final String ownerName) {
Preconditions.checkNotNull(ownerName);
lock.readLock().lock();
try {
@@ -402,7 +374,7 @@ public class ContainerStateMap {
* @param type - Replication type -- StandAlone, Ratis etc.
* @return NavigableSet
*/
- NavigableSet<ContainerID> getContainerIDsByType(ReplicationType type) {
+ NavigableSet<ContainerID> getContainerIDsByType(final ReplicationType type) {
Preconditions.checkNotNull(type);
lock.readLock().lock();
try {
@@ -418,7 +390,8 @@ public class ContainerStateMap {
* @param factor - Replication Factor.
* @return NavigableSet.
*/
- NavigableSet<ContainerID> getContainerIDsByFactor(ReplicationFactor factor) {
+ NavigableSet<ContainerID> getContainerIDsByFactor(
+ final ReplicationFactor factor) {
Preconditions.checkNotNull(factor);
lock.readLock().lock();
try {
@@ -435,7 +408,7 @@ public class ContainerStateMap {
* @return List of containers by state.
*/
public NavigableSet<ContainerID> getContainerIDsByState(
- LifeCycleState state) {
+ final LifeCycleState state) {
Preconditions.checkNotNull(state);
lock.readLock().lock();
try {
@@ -455,8 +428,8 @@ public class ContainerStateMap {
* @return ContainerInfo or Null if not container satisfies the criteria.
*/
public NavigableSet<ContainerID> getMatchingContainerIDs(
- LifeCycleState state, String owner,
- ReplicationFactor factor, ReplicationType type) {
+ final LifeCycleState state, final String owner,
+ final ReplicationFactor factor, final ReplicationType type) {
Preconditions.checkNotNull(state, "State cannot be null");
Preconditions.checkNotNull(owner, "Owner cannot be null");
@@ -465,7 +438,7 @@ public class ContainerStateMap {
lock.readLock().lock();
try {
- ContainerQueryKey queryKey =
+ final ContainerQueryKey queryKey =
new ContainerQueryKey(state, owner, factor, type);
if(resultCache.containsKey(queryKey)){
return resultCache.get(queryKey);
@@ -474,30 +447,33 @@ public class ContainerStateMap {
// If we cannot meet any one condition we return EMPTY_SET immediately.
// Since when we intersect these sets, the result will be empty if any
// one is empty.
- NavigableSet<ContainerID> stateSet =
+ final NavigableSet<ContainerID> stateSet =
lifeCycleStateMap.getCollection(state);
if (stateSet.size() == 0) {
return EMPTY_SET;
}
- NavigableSet<ContainerID> ownerSet = ownerMap.getCollection(owner);
+ final NavigableSet<ContainerID> ownerSet =
+ ownerMap.getCollection(owner);
if (ownerSet.size() == 0) {
return EMPTY_SET;
}
- NavigableSet<ContainerID> factorSet = factorMap.getCollection(factor);
+ final NavigableSet<ContainerID> factorSet =
+ factorMap.getCollection(factor);
if (factorSet.size() == 0) {
return EMPTY_SET;
}
- NavigableSet<ContainerID> typeSet = typeMap.getCollection(type);
+ final NavigableSet<ContainerID> typeSet =
+ typeMap.getCollection(type);
if (typeSet.size() == 0) {
return EMPTY_SET;
}
// if we add more constraints we will just add those sets here..
- NavigableSet<ContainerID>[] sets = sortBySize(stateSet,
+ final NavigableSet<ContainerID>[] sets = sortBySize(stateSet,
ownerSet, factorSet, typeSet);
NavigableSet<ContainerID> currentSet = sets[0];
@@ -521,12 +497,12 @@ public class ContainerStateMap {
* @return resultSet which is the intersection of these two sets.
*/
private NavigableSet<ContainerID> intersectSets(
- NavigableSet<ContainerID> smaller,
- NavigableSet<ContainerID> bigger) {
+ final NavigableSet<ContainerID> smaller,
+ final NavigableSet<ContainerID> bigger) {
Preconditions.checkState(smaller.size() <= bigger.size(),
"This function assumes the first set is lesser or equal to second " +
"set");
- NavigableSet<ContainerID> resultSet = new TreeSet<>();
+ final NavigableSet<ContainerID> resultSet = new TreeSet<>();
for (ContainerID id : smaller) {
if (bigger.contains(id)) {
resultSet.add(id);
@@ -544,11 +520,11 @@ public class ContainerStateMap {
*/
@SuppressWarnings("unchecked")
private NavigableSet<ContainerID>[] sortBySize(
- NavigableSet<ContainerID>... sets) {
+ final NavigableSet<ContainerID>... sets) {
for (int x = 0; x < sets.length - 1; x++) {
for (int y = 0; y < sets.length - x - 1; y++) {
if (sets[y].size() > sets[y + 1].size()) {
- NavigableSet temp = sets[y];
+ final NavigableSet temp = sets[y];
sets[y] = sets[y + 1];
sets[y + 1] = temp;
}
@@ -557,13 +533,22 @@ public class ContainerStateMap {
return sets;
}
- private void flushCache(ContainerInfo... containerInfos) {
+ private void flushCache(final ContainerInfo... containerInfos) {
for (ContainerInfo containerInfo : containerInfos) {
- ContainerQueryKey key = new ContainerQueryKey(containerInfo.getState(),
- containerInfo.getOwner(), containerInfo.getReplicationFactor(),
+ final ContainerQueryKey key = new ContainerQueryKey(
+ containerInfo.getState(),
+ containerInfo.getOwner(),
+ containerInfo.getReplicationFactor(),
containerInfo.getReplicationType());
resultCache.remove(key);
}
}
+ private void checkIfContainerExist(ContainerID containerID)
+ throws ContainerNotFoundException {
+ if (!containerMap.containsKey(containerID)) {
+ throw new ContainerNotFoundException("#" + containerID.getId());
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/50715c06/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DeadNodeHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DeadNodeHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DeadNodeHandler.java
index 17edf9e..1030428 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DeadNodeHandler.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DeadNodeHandler.java
@@ -21,11 +21,14 @@ package org.apache.hadoop.hdds.scm.node;
import java.util.Set;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.container.ContainerException;
import org.apache.hadoop.hdds.scm.container.ContainerID;
-import org.apache.hadoop.hdds.scm.container.ContainerStateManager;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerManager;
+import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException;
+import org.apache.hadoop.hdds.scm.container.ContainerReplica;
import org.apache.hadoop.hdds.scm.container.replication.ReplicationRequest;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
-import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.server.events.EventHandler;
import org.apache.hadoop.hdds.server.events.EventPublisher;
@@ -37,7 +40,7 @@ import org.slf4j.LoggerFactory;
*/
public class DeadNodeHandler implements EventHandler<DatanodeDetails> {
- private final ContainerStateManager containerStateManager;
+ private final ContainerManager containerManager;
private final NodeManager nodeManager;
@@ -45,8 +48,8 @@ public class DeadNodeHandler implements EventHandler<DatanodeDetails> {
LoggerFactory.getLogger(DeadNodeHandler.class);
public DeadNodeHandler(NodeManager nodeManager,
- ContainerStateManager containerStateManager) {
- this.containerStateManager = containerStateManager;
+ ContainerManager containerManager) {
+ this.containerManager = containerManager;
this.nodeManager = nodeManager;
}
@@ -55,45 +58,58 @@ public class DeadNodeHandler implements EventHandler<DatanodeDetails> {
EventPublisher publisher) {
nodeManager.processDeadNode(datanodeDetails.getUuid());
- Set<ContainerID> containers =
+ // TODO: check if there are any pipeline on this node and fire close
+ // pipeline event
+ Set<ContainerID> ids =
nodeManager.getContainers(datanodeDetails.getUuid());
- if (containers == null) {
+ if (ids == null) {
LOG.info("There's no containers in dead datanode {}, no replica will be"
+ " removed from the in-memory state.", datanodeDetails.getUuid());
return;
}
- LOG.info(
- "Datanode {} is dead. Removing replications from the in-memory state.",
- datanodeDetails.getUuid());
- for (ContainerID container : containers) {
+ LOG.info("Datanode {} is dead. Removing replications from the in-memory" +
+ " state.", datanodeDetails.getUuid());
+ for (ContainerID id : ids) {
try {
- try {
- containerStateManager.removeContainerReplica(container,
- datanodeDetails);
- } catch (SCMException ex) {
- LOG.info("DataNode {} doesn't have replica for container {}.",
- datanodeDetails.getUuid(), container.getId());
- }
-
- if (!containerStateManager.isOpen(container)) {
- ReplicationRequest replicationRequest =
- containerStateManager.checkReplicationState(container);
-
- if (replicationRequest != null) {
- publisher.fireEvent(SCMEvents.REPLICATE_CONTAINER,
- replicationRequest);
+ final ContainerInfo container = containerManager.getContainer(id);
+ if (!container.isOpen()) {
+ final ContainerReplica replica = ContainerReplica.newBuilder()
+ .setContainerID(id)
+ .setDatanodeDetails(datanodeDetails)
+ .build();
+ try {
+ containerManager.removeContainerReplica(id, replica);
+ replicateIfNeeded(container, publisher);
+ } catch (ContainerException ex) {
+ LOG.warn("Exception while removing container replica #{} for " +
+ "container #{}.", replica, container, ex);
}
}
- } catch (SCMException e) {
- LOG.error("Can't remove container from containerStateMap {}", container
- .getId(), e);
+ } catch (ContainerNotFoundException cnfe) {
+ LOG.warn("Container Not found!", cnfe);
}
}
}
/**
+ * Compare the existing replication number with the expected one.
+ */
+ private void replicateIfNeeded(ContainerInfo container,
+ EventPublisher publisher) throws ContainerNotFoundException {
+ final int existingReplicas = containerManager
+ .getContainerReplicas(container.containerID()).size();
+ final int expectedReplicas = container.getReplicationFactor().getNumber();
+ if (existingReplicas != expectedReplicas) {
+ publisher.fireEvent(SCMEvents.REPLICATE_CONTAINER,
+ new ReplicationRequest(
+ container.getContainerID(), existingReplicas, expectedReplicas));
+ }
+ }
+
+ /**
* Returns logger.
* */
+ // TODO: remove this.
public static Logger getLogger() {
return LOG;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/50715c06/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMChillModeManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMChillModeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMChillModeManager.java
index 44744f0..7135267 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMChillModeManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMChillModeManager.java
@@ -31,7 +31,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ScmOps;
-import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.server.SCMDatanodeProtocolServer
.NodeRegistrationContainerReport;
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org