You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by sh...@apache.org on 2021/03/04 12:03:03 UTC
[ozone] branch HDDS-2823 updated: HDDS-4894. Use
PipelineManagerV2Impl in Recon and enable ignored Recon test cases. (#1987)
This is an automated email from the ASF dual-hosted git repository.
shashikant pushed a commit to branch HDDS-2823
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/HDDS-2823 by this push:
new 00c52fb HDDS-4894. Use PipelineManagerV2Impl in Recon and enable ignored Recon test cases. (#1987)
00c52fb is described below
commit 00c52fb1fb6dc45adb3d02f3f9f82b6c8ad43fa7
Author: Glen Geng <gl...@tencent.com>
AuthorDate: Thu Mar 4 20:02:45 2021 +0800
HDDS-4894. Use PipelineManagerV2Impl in Recon and enable ignored Recon test cases. (#1987)
---
.../hadoop/hdds/scm/ha/SCMServiceManager.java | 2 +-
.../hdds/scm/pipeline/PipelineManagerV2Impl.java | 27 ++++---
.../hadoop/ozone/recon/TestReconAsPassiveScm.java | 17 ++---
.../ozone/recon/api/ClusterStateEndpoint.java | 8 +-
.../hadoop/ozone/recon/api/ContainerEndpoint.java | 7 +-
.../ozone/recon/scm/ReconContainerManager.java | 10 +--
.../ozone/recon/scm/ReconPipelineManager.java | 85 ++++++++++++++--------
.../scm/ReconStorageContainerManagerFacade.java | 24 ++++--
.../ozone/recon/fsck/TestContainerHealthTask.java | 6 +-
.../scm/AbstractReconContainerManagerTest.java | 22 +++++-
...TestReconIncrementalContainerReportHandler.java | 7 +-
.../ozone/recon/scm/TestReconPipelineManager.java | 39 ++++++++--
12 files changed, 163 insertions(+), 91 deletions(-)
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMServiceManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMServiceManager.java
index 14ece30..4fbd811 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMServiceManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMServiceManager.java
@@ -59,7 +59,7 @@ public final class SCMServiceManager {
*/
public synchronized void notifyEventTriggered(Event event) {
for (SCMService service : services) {
- LOG.debug("Notify service:{} with event:{.",
+ LOG.debug("Notify service:{} with event:{}.",
service.getServiceName(), event);
service.notifyEventTriggered(event);
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerV2Impl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerV2Impl.java
index a8ec6e4..eb0c948 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerV2Impl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerV2Impl.java
@@ -60,9 +60,9 @@ import java.util.concurrent.locks.ReentrantLock;
* All the write operations for pipelines must come via PipelineManager.
* It synchronises all write and read operations via a ReadWriteLock.
*/
-public final class PipelineManagerV2Impl implements PipelineManager {
+public class PipelineManagerV2Impl implements PipelineManager {
private static final Logger LOG =
- LoggerFactory.getLogger(SCMPipelineManager.class);
+ LoggerFactory.getLogger(PipelineManagerV2Impl.class);
// Limit the number of on-going ratis operation to be 1.
private final Lock lock;
@@ -79,13 +79,13 @@ public final class PipelineManagerV2Impl implements PipelineManager {
private final SCMContext scmContext;
private final NodeManager nodeManager;
- private PipelineManagerV2Impl(ConfigurationSource conf,
- SCMHAManager scmhaManager,
- NodeManager nodeManager,
- StateManager pipelineStateManager,
- PipelineFactory pipelineFactory,
- EventPublisher eventPublisher,
- SCMContext scmContext) {
+ protected PipelineManagerV2Impl(ConfigurationSource conf,
+ SCMHAManager scmhaManager,
+ NodeManager nodeManager,
+ StateManager pipelineStateManager,
+ PipelineFactory pipelineFactory,
+ EventPublisher eventPublisher,
+ SCMContext scmContext) {
this.lock = new ReentrantLock();
this.pipelineFactory = pipelineFactory;
this.stateManager = pipelineStateManager;
@@ -560,6 +560,11 @@ public final class PipelineManagerV2Impl implements PipelineManager {
return this.backgroundPipelineCreator;
}
+ @VisibleForTesting
+ public PipelineFactory getPipelineFactory() {
+ return pipelineFactory;
+ }
+
private void recordMetricsForPipeline(Pipeline pipeline) {
metrics.incNumPipelineAllocated();
if (pipeline.isOpen()) {
@@ -593,4 +598,8 @@ public final class PipelineManagerV2Impl implements PipelineManager {
return;
}
}
+
+ protected Lock getLock() {
+ return lock;
+ }
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/recon/TestReconAsPassiveScm.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/recon/TestReconAsPassiveScm.java
index efde033..cd27e1d 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/recon/TestReconAsPassiveScm.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/recon/TestReconAsPassiveScm.java
@@ -21,6 +21,7 @@ import java.util.Optional;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.scm.XceiverClientGrpc;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import org.apache.hadoop.hdds.scm.container.ContainerManagerV2;
import org.apache.hadoop.hdds.scm.node.NodeManager;
@@ -37,7 +38,6 @@ import org.apache.hadoop.test.LambdaTestUtils;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
@@ -90,7 +90,6 @@ public class TestReconAsPassiveScm {
}
@Test
- @Ignore
public void testDatanodeRegistrationAndReports() throws Exception {
ReconStorageContainerManagerFacade reconScm =
(ReconStorageContainerManagerFacade)
@@ -126,6 +125,7 @@ public class TestReconAsPassiveScm {
reconNodeManager.getAllNodes().size());
// Create container
+ ContainerManagerV2 reconContainerManager = reconScm.getContainerManager();
ContainerInfo containerInfo =
scmContainerManager.allocateContainer(RATIS, ONE, "test");
long containerID = containerInfo.getContainerID();
@@ -135,9 +135,8 @@ public class TestReconAsPassiveScm {
runTestOzoneContainerViaDataNode(containerID, client);
// Verify Recon picked up the new container that was created.
- // TODO: Fix ME
- //assertEquals(scmContainerManager.getContainerIDs(),
- // reconContainerManager.getContainerIDs());
+ assertEquals(scmContainerManager.getContainerIDs(),
+ reconContainerManager.getContainerIDs());
GenericTestUtils.LogCapturer logCapturer =
GenericTestUtils.LogCapturer.captureLogs(ReconNodeManager.LOG);
@@ -150,7 +149,6 @@ public class TestReconAsPassiveScm {
}
@Test
- @Ignore
public void testReconRestart() throws Exception {
final OzoneStorageContainerManager reconScm =
cluster.getReconServer().getReconStorageContainerManager();
@@ -204,9 +202,8 @@ public class TestReconAsPassiveScm {
assertFalse(
reconPipelineManager.containsPipeline(pipelineToClose.get().getId()));
- // TODO: Fix ME
- //LambdaTestUtils.await(90000, 5000,
- // () -> (newReconScm.getContainerManager()
- // .exists(ContainerID.valueOf(containerID))));
+ LambdaTestUtils.await(90000, 5000,
+ () -> (newReconScm.getContainerManager()
+ .containerExist(ContainerID.valueOf(containerID))));
}
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/ClusterStateEndpoint.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/ClusterStateEndpoint.java
index 0fe4af8..2123457 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/ClusterStateEndpoint.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/ClusterStateEndpoint.java
@@ -19,12 +19,12 @@
package org.apache.hadoop.ozone.recon.api;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
-import org.apache.hadoop.hdds.scm.container.ContainerManagerV2;
import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeStat;
import org.apache.hadoop.hdds.scm.node.NodeStatus;
import org.apache.hadoop.hdds.scm.server.OzoneStorageContainerManager;
import org.apache.hadoop.ozone.recon.api.types.ClusterStateResponse;
import org.apache.hadoop.ozone.recon.api.types.DatanodeStorageReport;
+import org.apache.hadoop.ozone.recon.scm.ReconContainerManager;
import org.apache.hadoop.ozone.recon.scm.ReconNodeManager;
import org.apache.hadoop.ozone.recon.scm.ReconPipelineManager;
import org.apache.hadoop.ozone.recon.tasks.TableCountTask;
@@ -57,9 +57,8 @@ public class ClusterStateEndpoint {
private ReconNodeManager nodeManager;
private ReconPipelineManager pipelineManager;
+ private ReconContainerManager containerManager;
private GlobalStatsDao globalStatsDao;
- // TODO: Fix ME, This has to be ReconContainerManager
- private ContainerManagerV2 containerManager;
@Inject
ClusterStateEndpoint(OzoneStorageContainerManager reconSCM,
@@ -67,8 +66,9 @@ public class ClusterStateEndpoint {
this.nodeManager =
(ReconNodeManager) reconSCM.getScmNodeManager();
this.pipelineManager = (ReconPipelineManager) reconSCM.getPipelineManager();
+ this.containerManager =
+ (ReconContainerManager) reconSCM.getContainerManager();
this.globalStatsDao = globalStatsDao;
- this.containerManager = reconSCM.getContainerManager();
}
/**
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/ContainerEndpoint.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/ContainerEndpoint.java
index 9ed0d44..0e26afe 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/ContainerEndpoint.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/ContainerEndpoint.java
@@ -84,15 +84,14 @@ public class ContainerEndpoint {
@Inject
private ReconOMMetadataManager omMetadataManager;
- // TODO: Fix ME, This has to be ReconContainerManager
- private ReconContainerManager containerManager;
+ private final ReconContainerManager containerManager;
private final ContainerHealthSchemaManager containerHealthSchemaManager;
@Inject
public ContainerEndpoint(OzoneStorageContainerManager reconSCM,
ContainerHealthSchemaManager containerHealthSchemaManager) {
- this.containerManager = (ReconContainerManager) reconSCM
- .getContainerManager();
+ this.containerManager =
+ (ReconContainerManager) reconSCM.getContainerManager();
this.containerHealthSchemaManager = containerHealthSchemaManager;
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconContainerManager.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconContainerManager.java
index 745d4fb..4d43672 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconContainerManager.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconContainerManager.java
@@ -42,8 +42,7 @@ import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException;
import org.apache.hadoop.hdds.scm.container.ContainerReplica;
import org.apache.hadoop.hdds.scm.container.ContainerReplicaNotFoundException;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
-import org.apache.hadoop.hdds.scm.ha.MockDBTransactionBuffer;
-import org.apache.hadoop.hdds.scm.ha.MockSCMHAManager;
+import org.apache.hadoop.hdds.scm.ha.SCMHAManager;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
import org.apache.hadoop.hdds.utils.db.DBStore;
@@ -82,6 +81,7 @@ public class ReconContainerManager extends ContainerManagerImpl {
*
* @throws IOException on Failure.
*/
+ @SuppressWarnings("parameternumber")
public ReconContainerManager(
Configuration conf,
DBStore store,
@@ -89,10 +89,10 @@ public class ReconContainerManager extends ContainerManagerImpl {
PipelineManager pipelineManager,
StorageContainerServiceProvider scm,
ContainerHealthSchemaManager containerHealthSchemaManager,
- ContainerDBServiceProvider containerDBServiceProvider)
+ ContainerDBServiceProvider containerDBServiceProvider,
+ SCMHAManager scmhaManager)
throws IOException {
- super(conf, MockSCMHAManager.getInstance(true,
- new MockDBTransactionBuffer(store)), pipelineManager, containerStore);
+ super(conf, scmhaManager, pipelineManager, containerStore);
this.scmClient = scm;
this.pipelineManager = pipelineManager;
this.containerHealthSchemaManager = containerHealthSchemaManager;
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconPipelineManager.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconPipelineManager.java
index 65b27a8..72e0a76 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconPipelineManager.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconPipelineManager.java
@@ -23,43 +23,68 @@ import java.util.List;
import java.util.stream.Collectors;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.ha.SCMContext;
+import org.apache.hadoop.hdds.scm.ha.SCMHAManager;
import org.apache.hadoop.hdds.scm.node.NodeManager;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineFactory;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
-import org.apache.hadoop.hdds.scm.pipeline.PipelineStateManager;
-import org.apache.hadoop.hdds.scm.pipeline.SCMPipelineManager;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineManagerV2Impl;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineStateManagerV2Impl;
+import org.apache.hadoop.hdds.scm.pipeline.StateManager;
import org.apache.hadoop.hdds.server.events.EventPublisher;
import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.ozone.ClientVersions;
import com.google.common.annotations.VisibleForTesting;
-import static org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState.CLOSED;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
+import static org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState.CLOSED;
/**
* Recon's overriding implementation of SCM's Pipeline Manager.
*/
-public class ReconPipelineManager extends SCMPipelineManager {
+public final class ReconPipelineManager extends PipelineManagerV2Impl {
private static final Logger LOG =
LoggerFactory.getLogger(ReconPipelineManager.class);
- public ReconPipelineManager(ConfigurationSource conf,
+ private ReconPipelineManager(ConfigurationSource conf,
+ SCMHAManager scmhaManager,
+ NodeManager nodeManager,
+ StateManager pipelineStateManager,
+ PipelineFactory pipelineFactory,
+ EventPublisher eventPublisher,
+ SCMContext scmContext) {
+ super(conf, scmhaManager, nodeManager, pipelineStateManager,
+ pipelineFactory, eventPublisher, scmContext);
+ }
+
+ public static ReconPipelineManager newReconPipelineManager(
+ ConfigurationSource conf,
NodeManager nodeManager,
Table<PipelineID, Pipeline> pipelineStore,
- EventPublisher eventPublisher)
- throws IOException {
- super(conf, nodeManager, pipelineStore, eventPublisher,
- new PipelineStateManager(),
- new ReconPipelineFactory());
- initializePipelineState();
- }
-
- @Override
- public void triggerPipelineCreation() {
- // Don't do anything in Recon.
+ EventPublisher eventPublisher,
+ SCMHAManager scmhaManager,
+ SCMContext scmContext) throws IOException {
+
+ // Create PipelineStateManager
+ StateManager stateManager = PipelineStateManagerV2Impl
+ .newBuilder()
+ .setPipelineStore(pipelineStore)
+ .setNodeManager(nodeManager)
+ .setRatisServer(scmhaManager.getRatisServer())
+ .setSCMDBTransactionBuffer(scmhaManager.getDBTransactionBuffer())
+ .build();
+
+ // Create PipelineFactory
+ PipelineFactory pipelineFactory = new ReconPipelineFactory();
+
+ return new ReconPipelineManager(conf, scmhaManager, nodeManager,
+ stateManager, pipelineFactory, eventPublisher, scmContext);
}
/**
@@ -69,7 +94,7 @@ public class ReconPipelineManager extends SCMPipelineManager {
*/
void initializePipelines(List<Pipeline> pipelinesFromScm) throws IOException {
- getLock().writeLock().lock();
+ getLock().lock();
try {
List<Pipeline> pipelinesInHouse = getPipelines();
LOG.info("Recon has {} pipelines in house.", pipelinesInHouse.size());
@@ -82,20 +107,21 @@ public class ReconPipelineManager extends SCMPipelineManager {
} else {
// Recon already has this pipeline. Just update state and creation
// time.
- getStateManager().updatePipelineState(pipeline.getId(),
- pipeline.getPipelineState());
+ getStateManager().updatePipelineState(
+ pipeline.getId().getProtobuf(),
+ Pipeline.PipelineState.getProtobuf(pipeline.getPipelineState()));
getPipeline(pipeline.getId()).setCreationTimestamp(
pipeline.getCreationTimestamp());
}
removeInvalidPipelines(pipelinesFromScm);
}
} finally {
- getLock().writeLock().unlock();
+ getLock().unlock();
}
}
public void removeInvalidPipelines(List<Pipeline> pipelinesFromScm) {
- getLock().writeLock().lock();
+ getLock().lock();
try {
List<Pipeline> pipelinesInHouse = getPipelines();
// Removing pipelines in Recon that are no longer in SCM.
@@ -109,7 +135,9 @@ public class ReconPipelineManager extends SCMPipelineManager {
PipelineID pipelineID = p.getId();
if (!p.getPipelineState().equals(CLOSED)) {
try {
- getStateManager().updatePipelineState(pipelineID, CLOSED);
+ getStateManager().updatePipelineState(
+ pipelineID.getProtobuf(),
+ HddsProtos.PipelineState.PIPELINE_CLOSED);
} catch (IOException e) {
LOG.warn("Pipeline {} not found while updating state. ",
p.getId(), e);
@@ -123,7 +151,7 @@ public class ReconPipelineManager extends SCMPipelineManager {
}
});
} finally {
- getLock().writeLock().unlock();
+ getLock().unlock();
}
}
/**
@@ -133,13 +161,12 @@ public class ReconPipelineManager extends SCMPipelineManager {
*/
@VisibleForTesting
public void addPipeline(Pipeline pipeline) throws IOException {
- getLock().writeLock().lock();
+ getLock().lock();
try {
- getPipelineStore().put(pipeline.getId(), pipeline);
- getStateManager().addPipeline(pipeline);
- getNodeManager().addPipeline(pipeline);
+ getStateManager().addPipeline(
+ pipeline.getProtobufMessage(ClientVersions.CURRENT_VERSION));
} finally {
- getLock().writeLock().unlock();
+ getLock().unlock();
}
}
}
diff --git a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStorageContainerManagerFacade.java b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStorageContainerManagerFacade.java
index 61aa494..ad05cb0 100644
--- a/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStorageContainerManagerFacade.java
+++ b/hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/scm/ReconStorageContainerManagerFacade.java
@@ -37,9 +37,12 @@ import org.apache.hadoop.hdds.scm.container.ReplicationManager;
import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementPolicyFactory;
import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementMetrics;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
+import org.apache.hadoop.hdds.scm.ha.MockDBTransactionBuffer;
+import org.apache.hadoop.hdds.scm.ha.MockSCMHAManager;
import org.apache.hadoop.hdds.scm.ha.SCMContext;
import org.apache.hadoop.hdds.scm.ha.SCMHANodeDetails;
import org.apache.hadoop.hdds.scm.ha.SCMNodeDetails;
+import org.apache.hadoop.hdds.scm.ha.SCMHAManager;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
import org.apache.hadoop.hdds.scm.net.NetworkTopologyImpl;
import org.apache.hadoop.hdds.scm.node.DeadNodeHandler;
@@ -86,6 +89,7 @@ public class ReconStorageContainerManagerFacade
private final SCMStorageConfig scmStorageConfig;
private final DBStore dbStore;
private final SCMHANodeDetails scmHANodeDetails;
+ private final SCMHAManager scmhaManager;
private ReconNodeManager nodeManager;
private ReconPipelineManager pipelineManager;
@@ -110,9 +114,10 @@ public class ReconStorageContainerManagerFacade
this.ozoneConfiguration = getReconScmConfiguration(conf);
this.scmStorageConfig = new ReconStorageConfig(conf);
this.clusterMap = new NetworkTopologyImpl(conf);
- dbStore = DBStoreBuilder
+ this.dbStore = DBStoreBuilder
.createDBStore(ozoneConfiguration, new ReconSCMDBDefinition());
-
+ this.scmhaManager = MockSCMHAManager.getInstance(
+ true, new MockDBTransactionBuffer(dbStore));
this.nodeManager =
new ReconNodeManager(conf, scmStorageConfig, eventQueue, clusterMap,
ReconSCMDBDefinition.NODES.getTable(dbStore));
@@ -122,16 +127,19 @@ public class ReconStorageContainerManagerFacade
clusterMap, true, placementMetrics);
this.datanodeProtocolServer = new ReconDatanodeProtocolServer(
conf, this, eventQueue);
- this.pipelineManager =
- new ReconPipelineManager(conf,
- nodeManager,
- ReconSCMDBDefinition.PIPELINES.getTable(dbStore),
- eventQueue);
+ this.pipelineManager = ReconPipelineManager.newReconPipelineManager(
+ conf,
+ nodeManager,
+ ReconSCMDBDefinition.PIPELINES.getTable(dbStore),
+ eventQueue,
+ scmhaManager,
+ scmContext);
this.containerManager = new ReconContainerManager(conf,
dbStore,
ReconSCMDBDefinition.CONTAINERS.getTable(dbStore),
pipelineManager, scmServiceProvider,
- containerHealthSchemaManager, containerDBServiceProvider);
+ containerHealthSchemaManager, containerDBServiceProvider,
+ scmhaManager);
this.scmServiceProvider = scmServiceProvider;
NodeReportHandler nodeReportHandler =
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/fsck/TestContainerHealthTask.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/fsck/TestContainerHealthTask.java
index 8de2a8d..dbbc2ff 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/fsck/TestContainerHealthTask.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/fsck/TestContainerHealthTask.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.ozone.recon.fsck;
import static junit.framework.TestCase.assertEquals;
import static org.junit.Assert.assertNotNull;
-import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -55,7 +54,6 @@ import org.hadoop.ozone.recon.schema.tables.daos.UnhealthyContainersDao;
import org.hadoop.ozone.recon.schema.tables.pojos.ReconTaskStatus;
import org.hadoop.ozone.recon.schema.tables.pojos.UnhealthyContainers;
import org.junit.Assert;
-import org.junit.Ignore;
import org.junit.Test;
/**
@@ -63,7 +61,6 @@ import org.junit.Test;
*/
public class TestContainerHealthTask extends AbstractReconSqlDBTest {
@Test
- @Ignore
public void testRun() throws Exception {
UnhealthyContainersDao unHealthyContainersTableHandle =
getDao(UnhealthyContainersDao.class);
@@ -85,8 +82,7 @@ public class TestContainerHealthTask extends AbstractReconSqlDBTest {
// defined below. The container with ID=6 will be healthy.
List<ContainerInfo> mockContainers = getMockContainers(6);
when(scmMock.getContainerManager()).thenReturn(containerManagerMock);
- when(containerManagerMock.getContainers(any(), any()))
- .thenReturn(mockContainers);
+ when(containerManagerMock.getContainers()).thenReturn(mockContainers);
for (ContainerInfo c : mockContainers) {
when(containerManagerMock.getContainer(c.containerID())).thenReturn(c);
}
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/AbstractReconContainerManagerTest.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/AbstractReconContainerManagerTest.java
index 8e71d40..86f1667 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/AbstractReconContainerManagerTest.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/AbstractReconContainerManagerTest.java
@@ -25,7 +25,10 @@ import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
+import org.apache.hadoop.hdds.scm.ha.MockDBTransactionBuffer;
+import org.apache.hadoop.hdds.scm.ha.MockSCMHAManager;
import org.apache.hadoop.hdds.scm.ha.SCMContext;
+import org.apache.hadoop.hdds.scm.ha.SCMHAManager;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
import org.apache.hadoop.hdds.scm.net.NetworkTopologyImpl;
import org.apache.hadoop.hdds.scm.node.NodeManager;
@@ -67,6 +70,8 @@ public class AbstractReconContainerManagerTest {
private ReconPipelineManager pipelineManager;
private ReconContainerManager containerManager;
private DBStore store;
+ private SCMHAManager scmhaManager;
+ private SCMContext scmContext;
@Before
public void setUp() throws Exception {
@@ -75,13 +80,21 @@ public class AbstractReconContainerManagerTest {
temporaryFolder.newFolder().getAbsolutePath());
conf.set(OZONE_SCM_NAMES, "localhost");
store = DBStoreBuilder.createDBStore(conf, new ReconSCMDBDefinition());
+ scmhaManager = MockSCMHAManager.getInstance(
+ true, new MockDBTransactionBuffer(store));
+ scmContext = SCMContext.emptyContext();
scmStorageConfig = new ReconStorageConfig(conf);
NetworkTopology clusterMap = new NetworkTopologyImpl(conf);
EventQueue eventQueue = new EventQueue();
NodeManager nodeManager = new SCMNodeManager(conf, scmStorageConfig,
- eventQueue, clusterMap, SCMContext.emptyContext());
- pipelineManager = new ReconPipelineManager(conf, nodeManager,
- ReconSCMDBDefinition.PIPELINES.getTable(store), eventQueue);
+ eventQueue, clusterMap, scmContext);
+ pipelineManager = ReconPipelineManager.newReconPipelineManager(
+ conf,
+ nodeManager,
+ ReconSCMDBDefinition.PIPELINES.getTable(store),
+ eventQueue,
+ scmhaManager,
+ scmContext);
containerManager = new ReconContainerManager(
conf,
store,
@@ -89,7 +102,8 @@ public class AbstractReconContainerManagerTest {
pipelineManager,
getScmServiceProvider(),
mock(ContainerHealthSchemaManager.class),
- mock(ContainerDBServiceProvider.class));
+ mock(ContainerDBServiceProvider.class),
+ scmhaManager);
}
@After
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconIncrementalContainerReportHandler.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconIncrementalContainerReportHandler.java
index 0b0bce5..54f8220 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconIncrementalContainerReportHandler.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconIncrementalContainerReportHandler.java
@@ -52,7 +52,6 @@ import org.apache.hadoop.hdds.scm.server.SCMStorageConfig;
import org.apache.hadoop.hdds.server.events.EventPublisher;
import org.apache.hadoop.hdds.server.events.EventQueue;
import org.apache.hadoop.test.GenericTestUtils;
-import org.junit.Ignore;
import org.junit.Test;
/**
@@ -62,7 +61,6 @@ public class TestReconIncrementalContainerReportHandler
extends AbstractReconContainerManagerTest {
@Test
- @Ignore
public void testProcessICR() throws IOException, NodeNotFoundException {
ContainerID containerID = ContainerID.valueOf(100L);
@@ -91,7 +89,7 @@ public class TestReconIncrementalContainerReportHandler
ReconContainerManager containerManager = getContainerManager();
ReconIncrementalContainerReportHandler reconIcr =
new ReconIncrementalContainerReportHandler(nodeManager,
- null, SCMContext.emptyContext());
+ containerManager, SCMContext.emptyContext());
EventPublisher eventPublisherMock = mock(EventPublisher.class);
reconIcr.onMessage(reportMock, eventPublisherMock);
@@ -102,7 +100,6 @@ public class TestReconIncrementalContainerReportHandler
}
@Test
- @Ignore
public void testProcessICRStateMismatch() throws IOException {
// Recon container state is "OPEN".
@@ -134,7 +131,7 @@ public class TestReconIncrementalContainerReportHandler
when(reportMock.getReport()).thenReturn(containerReport);
ReconIncrementalContainerReportHandler reconIcr =
new ReconIncrementalContainerReportHandler(nodeManagerMock,
- null, SCMContext.emptyContext());
+ containerManager, SCMContext.emptyContext());
reconIcr.onMessage(reportMock, mock(EventPublisher.class));
assertTrue(containerManager.containerExist(containerID));
diff --git a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconPipelineManager.java b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconPipelineManager.java
index 2c4226f..1de004b 100644
--- a/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconPipelineManager.java
+++ b/hadoop-ozone/recon/src/test/java/org/apache/hadoop/ozone/recon/scm/TestReconPipelineManager.java
@@ -26,7 +26,10 @@ import java.util.List;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
+import org.apache.hadoop.hdds.scm.ha.MockDBTransactionBuffer;
+import org.apache.hadoop.hdds.scm.ha.MockSCMHAManager;
import org.apache.hadoop.hdds.scm.ha.SCMContext;
+import org.apache.hadoop.hdds.scm.ha.SCMHAManager;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
import org.apache.hadoop.hdds.scm.net.NetworkTopologyImpl;
import org.apache.hadoop.hdds.scm.node.NodeManager;
@@ -66,6 +69,8 @@ public class TestReconPipelineManager {
private OzoneConfiguration conf;
private SCMStorageConfig scmStorageConfig;
private DBStore store;
+ private SCMHAManager scmhaManager;
+ private SCMContext scmContext;
@Before
public void setup() throws IOException {
@@ -75,6 +80,9 @@ public class TestReconPipelineManager {
conf.set(OZONE_SCM_NAMES, "localhost");
scmStorageConfig = new ReconStorageConfig(conf);
store = DBStoreBuilder.createDBStore(conf, new ReconSCMDBDefinition());
+ scmhaManager = MockSCMHAManager.getInstance(
+ true, new MockDBTransactionBuffer(store));
+ scmContext = SCMContext.emptyContext();
}
@After
@@ -114,8 +122,13 @@ public class TestReconPipelineManager {
eventQueue, clusterMap, SCMContext.emptyContext());
try (ReconPipelineManager reconPipelineManager =
- new ReconPipelineManager(conf, nodeManager,
- ReconSCMDBDefinition.PIPELINES.getTable(store), eventQueue)) {
+ ReconPipelineManager.newReconPipelineManager(
+ conf,
+ nodeManager,
+ ReconSCMDBDefinition.PIPELINES.getTable(store),
+ eventQueue,
+ scmhaManager,
+ scmContext)) {
reconPipelineManager.addPipeline(validPipeline);
reconPipelineManager.addPipeline(invalidPipeline);
@@ -150,8 +163,14 @@ public class TestReconPipelineManager {
eventQueue, clusterMap, SCMContext.emptyContext());
ReconPipelineManager reconPipelineManager =
- new ReconPipelineManager(conf, nodeManager,
- ReconSCMDBDefinition.PIPELINES.getTable(store), eventQueue);
+ ReconPipelineManager.newReconPipelineManager(
+ conf,
+ nodeManager,
+ ReconSCMDBDefinition.PIPELINES.getTable(store),
+ eventQueue,
+ scmhaManager,
+ scmContext);
+
assertFalse(reconPipelineManager.containsPipeline(pipeline.getId()));
reconPipelineManager.addPipeline(pipeline);
assertTrue(reconPipelineManager.containsPipeline(pipeline.getId()));
@@ -162,9 +181,15 @@ public class TestReconPipelineManager {
NodeManager nodeManagerMock = mock(NodeManager.class);
- ReconPipelineManager reconPipelineManager = new ReconPipelineManager(
- conf, nodeManagerMock, ReconSCMDBDefinition.PIPELINES.getTable(store),
- new EventQueue());
+ ReconPipelineManager reconPipelineManager =
+ ReconPipelineManager.newReconPipelineManager(
+ conf,
+ nodeManagerMock,
+ ReconSCMDBDefinition.PIPELINES.getTable(store),
+ new EventQueue(),
+ scmhaManager,
+ scmContext);
+
PipelineFactory pipelineFactory = reconPipelineManager.getPipelineFactory();
assertTrue(pipelineFactory instanceof ReconPipelineFactory);
ReconPipelineFactory reconPipelineFactory =
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org