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 ae...@apache.org on 2018/06/14 03:31:48 UTC
hadoop git commit: HDDS-163. Add Datanode heartbeat dispatcher in
SCM. Contributed by Nandakumar.
Repository: hadoop
Updated Branches:
refs/heads/trunk 7547740e5 -> ddd09d59f
HDDS-163. Add Datanode heartbeat dispatcher in SCM.
Contributed by Nandakumar.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/ddd09d59
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/ddd09d59
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/ddd09d59
Branch: refs/heads/trunk
Commit: ddd09d59f3d9825f068026622720914e04c2e1d6
Parents: 7547740
Author: Anu Engineer <ae...@apache.org>
Authored: Wed Jun 13 20:18:22 2018 -0700
Committer: Anu Engineer <ae...@apache.org>
Committed: Wed Jun 13 20:18:22 2018 -0700
----------------------------------------------------------------------
.../container/common/report/ReportManager.java | 3 +-
.../scm/server/SCMDatanodeProtocolServer.java | 119 +++++-------
.../SCMDatanodeContainerReportHandler.java | 76 ++++++++
.../report/SCMDatanodeHeartbeatDispatcher.java | 189 +++++++++++++++++++
.../report/SCMDatanodeNodeReportHandler.java | 43 +++++
.../server/report/SCMDatanodeReportHandler.java | 83 ++++++++
.../report/SCMDatanodeReportHandlerFactory.java | 82 ++++++++
.../hdds/scm/server/report/package-info.java | 57 ++++++
.../TestSCMDatanodeContainerReportHandler.java | 34 ++++
.../TestSCMDatanodeHeartbeatDispatcher.java | 138 ++++++++++++++
.../TestSCMDatanodeNodeReportHandler.java | 36 ++++
.../TestSCMDatanodeReportHandlerFactory.java | 51 +++++
.../hdds/scm/server/report/package-info.java | 21 +++
.../apache/hadoop/ozone/scm/TestSCMMetrics.java | 20 +-
14 files changed, 875 insertions(+), 77 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/report/ReportManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/report/ReportManager.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/report/ReportManager.java
index c09282e..8097cd6 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/report/ReportManager.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/report/ReportManager.java
@@ -50,7 +50,8 @@ public final class ReportManager {
List<ReportPublisher> publishers) {
this.context = context;
this.publishers = publishers;
- this.executorService = HadoopExecutors.newScheduledThreadPool(1,
+ this.executorService = HadoopExecutors.newScheduledThreadPool(
+ publishers.size(),
new ThreadFactoryBuilder().setDaemon(true)
.setNameFormat("Datanode ReportManager Thread - %d").build());
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java
index 1b1645d..7d16161 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMDatanodeProtocolServer.java
@@ -69,7 +69,7 @@ import static org.apache.hadoop.hdds.protocol.proto
import org.apache.hadoop.hdds.scm.HddsServerUtil;
-import org.apache.hadoop.hdds.scm.container.placement.metrics.ContainerStat;
+import org.apache.hadoop.hdds.scm.server.report.SCMDatanodeHeartbeatDispatcher;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.ipc.ProtobufRpcEngine;
import org.apache.hadoop.ipc.RPC;
@@ -114,6 +114,7 @@ public class SCMDatanodeProtocolServer implements
private final StorageContainerManager scm;
private final InetSocketAddress datanodeRpcAddress;
+ private final SCMDatanodeHeartbeatDispatcher heartbeatDispatcher;
public SCMDatanodeProtocolServer(final OzoneConfiguration conf,
StorageContainerManager scm) throws IOException {
@@ -148,14 +149,22 @@ public class SCMDatanodeProtocolServer implements
updateRPCListenAddress(
conf, OZONE_SCM_DATANODE_ADDRESS_KEY, datanodeRpcAddr,
datanodeRpcServer);
+
+ heartbeatDispatcher = SCMDatanodeHeartbeatDispatcher.newBuilder(conf, scm)
+ .addHandlerFor(NodeReportProto.class)
+ .addHandlerFor(ContainerReportsProto.class)
+ .build();
}
- public InetSocketAddress getDatanodeRpcAddress() {
- return datanodeRpcAddress;
+ public void start() {
+ LOG.info(
+ StorageContainerManager.buildRpcServerStartMessage(
+ "RPC server for DataNodes", datanodeRpcAddress));
+ datanodeRpcServer.start();
}
- public RPC.Server getDatanodeRpcServer() {
- return datanodeRpcServer;
+ public InetSocketAddress getDatanodeRpcAddress() {
+ return datanodeRpcAddress;
}
@Override
@@ -167,25 +176,6 @@ public class SCMDatanodeProtocolServer implements
}
@Override
- public SCMHeartbeatResponseProto sendHeartbeat(
- SCMHeartbeatRequestProto heartbeat)
- throws IOException {
- // TODO: Add a heartbeat dispatcher.
- DatanodeDetails datanodeDetails = DatanodeDetails
- .getFromProtoBuf(heartbeat.getDatanodeDetails());
- NodeReportProto nodeReport = heartbeat.getNodeReport();
- List<SCMCommand> commands =
- scm.getScmNodeManager().sendHeartbeat(datanodeDetails, nodeReport);
- List<SCMCommandProto> cmdResponses = new LinkedList<>();
- for (SCMCommand cmd : commands) {
- cmdResponses.add(getCommandResponse(cmd));
- }
- return SCMHeartbeatResponseProto.newBuilder()
- .setDatanodeUUID(datanodeDetails.getUuidString())
- .addAllCommands(cmdResponses).build();
- }
-
- @Override
public SCMRegisteredResponseProto register(
HddsProtos.DatanodeDetailsProto datanodeDetailsProto,
NodeReportProto nodeReport,
@@ -216,36 +206,27 @@ public class SCMDatanodeProtocolServer implements
.build();
}
- public void processContainerReports(DatanodeDetails datanodeDetails,
- ContainerReportsProto reports)
+ @Override
+ public SCMHeartbeatResponseProto sendHeartbeat(
+ SCMHeartbeatRequestProto heartbeat)
throws IOException {
- updateContainerReportMetrics(datanodeDetails, reports);
- // should we process container reports async?
- scm.getScmContainerManager()
- .processContainerReports(datanodeDetails, reports);
- }
+ heartbeatDispatcher.dispatch(heartbeat);
- private void updateContainerReportMetrics(DatanodeDetails datanodeDetails,
- ContainerReportsProto reports) {
- ContainerStat newStat = new ContainerStat();
- for (StorageContainerDatanodeProtocolProtos.ContainerInfo info : reports
- .getReportsList()) {
- newStat.add(new ContainerStat(info.getSize(), info.getUsed(),
- info.getKeyCount(), info.getReadBytes(), info.getWriteBytes(),
- info.getReadCount(), info.getWriteCount()));
+ // TODO: Remove the below code after SCM refactoring.
+ DatanodeDetails datanodeDetails = DatanodeDetails
+ .getFromProtoBuf(heartbeat.getDatanodeDetails());
+ NodeReportProto nodeReport = heartbeat.getNodeReport();
+ List<SCMCommand> commands =
+ scm.getScmNodeManager().sendHeartbeat(datanodeDetails, nodeReport);
+ List<SCMCommandProto> cmdResponses = new LinkedList<>();
+ for (SCMCommand cmd : commands) {
+ cmdResponses.add(getCommandResponse(cmd));
}
- // update container metrics
- StorageContainerManager.getMetrics().setLastContainerStat(newStat);
-
- // Update container stat entry, this will trigger a removal operation if it
- // exists in cache.
- String datanodeUuid = datanodeDetails.getUuidString();
- scm.getContainerReportCache().put(datanodeUuid, newStat);
- // update global view container metrics
- StorageContainerManager.getMetrics().incrContainerStat(newStat);
+ return SCMHeartbeatResponseProto.newBuilder()
+ .setDatanodeUUID(datanodeDetails.getUuidString())
+ .addAllCommands(cmdResponses).build();
}
-
@Override
public ContainerBlocksDeletionACKResponseProto sendContainerBlocksDeletionACK(
ContainerBlocksDeletionACKProto acks) throws IOException {
@@ -271,28 +252,6 @@ public class SCMDatanodeProtocolServer implements
.getDefaultInstanceForType();
}
- public void start() {
- LOG.info(
- StorageContainerManager.buildRpcServerStartMessage(
- "RPC server for DataNodes", getDatanodeRpcAddress()));
- getDatanodeRpcServer().start();
- }
-
- public void stop() {
- try {
- LOG.info("Stopping the RPC server for DataNodes");
- datanodeRpcServer.stop();
- } catch (Exception ex) {
- LOG.error(" datanodeRpcServer stop failed.", ex);
- }
- IOUtils.cleanupWithLogger(LOG, scm.getScmNodeManager());
- }
-
- public void join() throws InterruptedException {
- LOG.trace("Join RPC server for DataNodes");
- datanodeRpcServer.join();
- }
-
/**
* Returns a SCMCommandRepose from the SCM Command.
*
@@ -338,4 +297,22 @@ public class SCMDatanodeProtocolServer implements
throw new IllegalArgumentException("Not implemented");
}
}
+
+
+ public void join() throws InterruptedException {
+ LOG.trace("Join RPC server for DataNodes");
+ datanodeRpcServer.join();
+ }
+
+ public void stop() {
+ try {
+ LOG.info("Stopping the RPC server for DataNodes");
+ datanodeRpcServer.stop();
+ heartbeatDispatcher.shutdown();
+ } catch (Exception ex) {
+ LOG.error(" datanodeRpcServer stop failed.", ex);
+ }
+ IOUtils.cleanupWithLogger(LOG, scm.getScmNodeManager());
+ }
+
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeContainerReportHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeContainerReportHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeContainerReportHandler.java
new file mode 100644
index 0000000..00ce94d
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeContainerReportHandler.java
@@ -0,0 +1,76 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.report;
+
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
+import org.apache.hadoop.hdds.scm.container.placement.metrics.ContainerStat;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * Handler for Datanode Container Report.
+ */
+public class SCMDatanodeContainerReportHandler extends
+ SCMDatanodeReportHandler<ContainerReportsProto> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ SCMDatanodeContainerReportHandler.class);
+
+ @Override
+ public void processReport(DatanodeDetails datanodeDetails,
+ ContainerReportsProto report) throws IOException {
+ LOG.trace("Processing container report from {}.", datanodeDetails);
+ updateContainerReportMetrics(datanodeDetails, report);
+ getSCM().getScmContainerManager()
+ .processContainerReports(datanodeDetails, report);
+ }
+
+ /**
+ * Updates container report metrics in SCM.
+ *
+ * @param datanodeDetails Datanode Information
+ * @param reports Container Reports
+ */
+ private void updateContainerReportMetrics(DatanodeDetails datanodeDetails,
+ ContainerReportsProto reports) {
+ ContainerStat newStat = new ContainerStat();
+ for (StorageContainerDatanodeProtocolProtos.ContainerInfo info : reports
+ .getReportsList()) {
+ newStat.add(new ContainerStat(info.getSize(), info.getUsed(),
+ info.getKeyCount(), info.getReadBytes(), info.getWriteBytes(),
+ info.getReadCount(), info.getWriteCount()));
+ }
+ // update container metrics
+ StorageContainerManager.getMetrics().setLastContainerStat(newStat);
+
+ // Update container stat entry, this will trigger a removal operation if it
+ // exists in cache.
+ String datanodeUuid = datanodeDetails.getUuidString();
+ getSCM().getContainerReportCache().put(datanodeUuid, newStat);
+ // update global view container metrics
+ StorageContainerManager.getMetrics().incrContainerStat(newStat);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeHeartbeatDispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeHeartbeatDispatcher.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeHeartbeatDispatcher.java
new file mode 100644
index 0000000..d50edff
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeHeartbeatDispatcher.java
@@ -0,0 +1,189 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.report;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import com.google.protobuf.GeneratedMessage;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.util.concurrent.HadoopExecutors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+
+/**
+ * This class is responsible for dispatching heartbeat from datanode to
+ * appropriate ReportHandlers at SCM.
+ * Only one handler per report is supported now, it's very easy to support
+ * multiple handlers for a report.
+ */
+public final class SCMDatanodeHeartbeatDispatcher {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ SCMDatanodeHeartbeatDispatcher.class);
+
+ /**
+ * This stores Report to Handler mapping.
+ */
+ private final Map<Class<? extends GeneratedMessage>,
+ SCMDatanodeReportHandler<? extends GeneratedMessage>> handlers;
+
+ /**
+ * Executor service which will be used for processing reports.
+ */
+ private final ExecutorService executorService;
+
+ /**
+ * Constructs SCMDatanodeHeartbeatDispatcher instance with the given
+ * handlers.
+ *
+ * @param handlers report to report handler mapping
+ */
+ private SCMDatanodeHeartbeatDispatcher(Map<Class<? extends GeneratedMessage>,
+ SCMDatanodeReportHandler<? extends GeneratedMessage>> handlers) {
+ this.handlers = handlers;
+ this.executorService = HadoopExecutors.newCachedThreadPool(
+ new ThreadFactoryBuilder().setDaemon(true)
+ .setNameFormat("SCMDatanode Heartbeat Dispatcher Thread - %d")
+ .build());
+ }
+
+ /**
+ * Dispatches heartbeat to registered handlers.
+ *
+ * @param heartbeat heartbeat to be dispatched.
+ */
+ public void dispatch(SCMHeartbeatRequestProto heartbeat) {
+ DatanodeDetails datanodeDetails = DatanodeDetails
+ .getFromProtoBuf(heartbeat.getDatanodeDetails());
+ if (heartbeat.hasNodeReport()) {
+ processReport(datanodeDetails, heartbeat.getNodeReport());
+ }
+ if (heartbeat.hasContainerReport()) {
+ processReport(datanodeDetails, heartbeat.getContainerReport());
+ }
+ }
+
+ /**
+ * Invokes appropriate ReportHandler and submits the task to executor
+ * service for processing.
+ *
+ * @param datanodeDetails Datanode Information
+ * @param report Report to be processed
+ */
+ @SuppressWarnings("unchecked")
+ private void processReport(DatanodeDetails datanodeDetails,
+ GeneratedMessage report) {
+ executorService.submit(() -> {
+ try {
+ SCMDatanodeReportHandler handler = handlers.get(report.getClass());
+ handler.processReport(datanodeDetails, report);
+ } catch (IOException ex) {
+ LOG.error("Exception wile processing report {}, from {}",
+ report.getClass(), datanodeDetails, ex);
+ }
+ });
+ }
+
+ /**
+ * Shuts down SCMDatanodeHeartbeatDispatcher.
+ */
+ public void shutdown() {
+ executorService.shutdown();
+ }
+
+ /**
+ * Returns a new Builder to construct {@link SCMDatanodeHeartbeatDispatcher}.
+ *
+ * @param conf Configuration to be used by SCMDatanodeHeartbeatDispatcher
+ * @param scm {@link StorageContainerManager} instance to be used by report
+ * handlers
+ *
+ * @return {@link SCMDatanodeHeartbeatDispatcher.Builder} instance
+ */
+ public static Builder newBuilder(Configuration conf,
+ StorageContainerManager scm) {
+ return new Builder(conf, scm);
+ }
+
+ /**
+ * Builder for SCMDatanodeHeartbeatDispatcher.
+ */
+ public static class Builder {
+
+ private final SCMDatanodeReportHandlerFactory reportHandlerFactory;
+ private final Map<Class<? extends GeneratedMessage>,
+ SCMDatanodeReportHandler<? extends GeneratedMessage>> report2handler;
+
+ /**
+ * Constructs SCMDatanodeHeartbeatDispatcher.Builder instance.
+ *
+ * @param conf Configuration object to be used.
+ * @param scm StorageContainerManager instance to be used for report
+ * handler initialization.
+ */
+ private Builder(Configuration conf, StorageContainerManager scm) {
+ this.report2handler = new HashMap<>();
+ this.reportHandlerFactory =
+ new SCMDatanodeReportHandlerFactory(conf, scm);
+ }
+
+ /**
+ * Adds new report handler for the given report.
+ *
+ * @param report Report for which handler has to be added
+ *
+ * @return Builder
+ */
+ public Builder addHandlerFor(Class<? extends GeneratedMessage> report) {
+ report2handler.put(report, reportHandlerFactory.getHandlerFor(report));
+ return this;
+ }
+
+ /**
+ * Associates the given report handler for the given report.
+ *
+ * @param report Report to be associated with
+ * @param handler Handler to be used for the report
+ *
+ * @return Builder
+ */
+ public Builder addHandler(Class<? extends GeneratedMessage> report,
+ SCMDatanodeReportHandler<? extends GeneratedMessage> handler) {
+ report2handler.put(report, handler);
+ return this;
+ }
+
+ /**
+ * Builds and returns {@link SCMDatanodeHeartbeatDispatcher} instance.
+ *
+ * @return SCMDatanodeHeartbeatDispatcher
+ */
+ public SCMDatanodeHeartbeatDispatcher build() {
+ return new SCMDatanodeHeartbeatDispatcher(report2handler);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeNodeReportHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeNodeReportHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeNodeReportHandler.java
new file mode 100644
index 0000000..fb89b02
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeNodeReportHandler.java
@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.report;
+
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.NodeReportProto;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * Handles Datanode Node Report.
+ */
+public class SCMDatanodeNodeReportHandler extends
+ SCMDatanodeReportHandler<NodeReportProto> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ SCMDatanodeNodeReportHandler.class);
+
+ @Override
+ public void processReport(DatanodeDetails datanodeDetails,
+ NodeReportProto report) throws IOException {
+ LOG.debug("Processing node report from {}.", datanodeDetails);
+ //TODO: add logic to process node report.
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeReportHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeReportHandler.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeReportHandler.java
new file mode 100644
index 0000000..d338649
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeReportHandler.java
@@ -0,0 +1,83 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.report;
+
+import com.google.protobuf.GeneratedMessage;
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+
+import java.io.IOException;
+
+/**
+ * Datanode Report handlers should implement this interface in order to get
+ * call back whenever the report is received from datanode.
+ *
+ * @param <T> Type of report the handler is interested in.
+ */
+public abstract class SCMDatanodeReportHandler<T extends GeneratedMessage>
+ implements Configurable {
+
+ private Configuration config;
+ private StorageContainerManager scm;
+
+ /**
+ * Initializes SCMDatanodeReportHandler and associates it with the given
+ * StorageContainerManager instance.
+ *
+ * @param storageContainerManager StorageContainerManager instance to be
+ * associated with.
+ */
+ public void init(StorageContainerManager storageContainerManager) {
+ this.scm = storageContainerManager;
+ }
+
+ /**
+ * Returns the associated StorageContainerManager instance. This will be
+ * used by the ReportHandler implementations.
+ *
+ * @return {@link StorageContainerManager}
+ */
+ protected StorageContainerManager getSCM() {
+ return scm;
+ }
+
+ @Override
+ public void setConf(Configuration conf) {
+ this.config = conf;
+ }
+
+ @Override
+ public Configuration getConf() {
+ return config;
+ }
+
+ /**
+ * Processes the report received from datanode. Each ReportHandler
+ * implementation is responsible for providing the logic to process the
+ * report it's interested in.
+ *
+ * @param datanodeDetails Datanode Information
+ * @param report Report to be processed
+ *
+ * @throws IOException In case of any exception
+ */
+ abstract void processReport(DatanodeDetails datanodeDetails, T report)
+ throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeReportHandlerFactory.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeReportHandlerFactory.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeReportHandlerFactory.java
new file mode 100644
index 0000000..e88495f
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/SCMDatanodeReportHandlerFactory.java
@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.report;
+
+import com.google.protobuf.GeneratedMessage;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.NodeReportProto;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.util.ReflectionUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+
+/**
+ * Factory class to construct {@link SCMDatanodeReportHandler} given a report.
+ */
+public class SCMDatanodeReportHandlerFactory {
+
+ private final Configuration conf;
+ private final StorageContainerManager scm;
+ private final Map<Class<? extends GeneratedMessage>,
+ Class<? extends SCMDatanodeReportHandler<? extends GeneratedMessage>>>
+ report2handler;
+
+ /**
+ * Constructs {@link SCMDatanodeReportHandler} instance.
+ *
+ * @param conf Configuration to be passed to the
+ * {@link SCMDatanodeReportHandler}
+ */
+ public SCMDatanodeReportHandlerFactory(Configuration conf,
+ StorageContainerManager scm) {
+ this.conf = conf;
+ this.scm = scm;
+ this.report2handler = new HashMap<>();
+
+ report2handler.put(NodeReportProto.class,
+ SCMDatanodeNodeReportHandler.class);
+ report2handler.put(ContainerReportsProto.class,
+ SCMDatanodeContainerReportHandler.class);
+ }
+
+ /**
+ * Returns the SCMDatanodeReportHandler for the corresponding report.
+ *
+ * @param report report
+ *
+ * @return report handler
+ */
+ public SCMDatanodeReportHandler<? extends GeneratedMessage> getHandlerFor(
+ Class<? extends GeneratedMessage> report) {
+ Class<? extends SCMDatanodeReportHandler<? extends GeneratedMessage>>
+ handlerClass = report2handler.get(report);
+ if (handlerClass == null) {
+ throw new RuntimeException("No handler found for report " + report);
+ }
+ SCMDatanodeReportHandler<? extends GeneratedMessage> instance =
+ ReflectionUtils.newInstance(handlerClass, conf);
+ instance.init(scm);
+ return instance;
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/package-info.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/package-info.java
new file mode 100644
index 0000000..fda3993
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/report/package-info.java
@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.hadoop.hdds.scm.server.report;
+/**
+ * Handling of all the datanode reports in SCM which are received through
+ * heartbeat is done here.
+ *
+ * SCM Datanode Report Processing State Diagram:
+ *
+ * SCMDatanode SCMDatanodeHeartbeat SCMDatanodeReport
+ * ProtocolServer Dispatcher Handler
+ * | | |
+ * | | |
+ * | construct | |
+ * |----------------------->| |
+ * | | |
+ * | | register |
+ * | |<-----------------------|
+ * | | |
+ * +------------+------------------------+------------------------+--------+
+ * | loop | | | |
+ * | | | | |
+ * | | | | |
+ * | heartbeat | | | |
+ * - +----------->| | | |
+ * | from | heartbeat | | |
+ * | Datanode |----------------------->| | |
+ * | | | report | |
+ * | | |----------------------->| |
+ * | | | | |
+ * | DN | | | |
+ * <-+------------| | | |
+ * | commands | | | |
+ * | | | | |
+ * +------------+------------------------+------------------------+--------+
+ * | | |
+ * | | |
+ * | shutdown | |
+ * |----------------------->| |
+ * | | |
+ * | | |
+ * - - -
+ */
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeContainerReportHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeContainerReportHandler.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeContainerReportHandler.java
new file mode 100644
index 0000000..776ae88
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeContainerReportHandler.java
@@ -0,0 +1,34 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.report;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Test cases to verify SCMDatanodeContainerReportHandler's behavior.
+ */
+public class TestSCMDatanodeContainerReportHandler {
+
+ //TODO: add test cases to verify SCMDatanodeContainerReportHandler.
+
+ @Test
+ public void dummyTest() {
+ Assert.assertTrue(true);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeHeartbeatDispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeHeartbeatDispatcher.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeHeartbeatDispatcher.java
new file mode 100644
index 0000000..5d08647
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeHeartbeatDispatcher.java
@@ -0,0 +1,138 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.report;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.NodeReportProto;
+import org.apache.hadoop.hdds.scm.TestUtils;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.io.IOException;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/**
+ * This class tests the behavior of SCMDatanodeHeartbeatDispatcher.
+ */
+public class TestSCMDatanodeHeartbeatDispatcher {
+
+ @Test
+ public void testSCMDatanodeHeartbeatDispatcherBuilder() {
+ Configuration conf = new OzoneConfiguration();
+ SCMDatanodeHeartbeatDispatcher dispatcher =
+ SCMDatanodeHeartbeatDispatcher.newBuilder(conf, null)
+ .addHandlerFor(NodeReportProto.class)
+ .addHandlerFor(ContainerReportsProto.class)
+ .build();
+ Assert.assertNotNull(dispatcher);
+ }
+
+ @Test
+ public void testNodeReportDispatcher() throws IOException {
+ Configuration conf = new OzoneConfiguration();
+ SCMDatanodeNodeReportHandler nodeReportHandler =
+ Mockito.mock(SCMDatanodeNodeReportHandler.class);
+ SCMDatanodeHeartbeatDispatcher dispatcher =
+ SCMDatanodeHeartbeatDispatcher.newBuilder(conf, null)
+ .addHandler(NodeReportProto.class, nodeReportHandler)
+ .build();
+
+ DatanodeDetails datanodeDetails = TestUtils.getDatanodeDetails();
+ NodeReportProto nodeReport = NodeReportProto.getDefaultInstance();
+ SCMHeartbeatRequestProto heartbeat =
+ SCMHeartbeatRequestProto.newBuilder()
+ .setDatanodeDetails(datanodeDetails.getProtoBufMessage())
+ .setNodeReport(nodeReport)
+ .build();
+ dispatcher.dispatch(heartbeat);
+ verify(nodeReportHandler,
+ times(1))
+ .processReport(any(DatanodeDetails.class), eq(nodeReport));
+ }
+
+ @Test
+ public void testContainerReportDispatcher() throws IOException {
+ Configuration conf = new OzoneConfiguration();
+ SCMDatanodeContainerReportHandler containerReportHandler =
+ Mockito.mock(SCMDatanodeContainerReportHandler.class);
+ SCMDatanodeHeartbeatDispatcher dispatcher =
+ SCMDatanodeHeartbeatDispatcher.newBuilder(conf, null)
+ .addHandler(ContainerReportsProto.class, containerReportHandler)
+ .build();
+
+ DatanodeDetails datanodeDetails = TestUtils.getDatanodeDetails();
+ ContainerReportsProto containerReport =
+ ContainerReportsProto.getDefaultInstance();
+ SCMHeartbeatRequestProto heartbeat =
+ SCMHeartbeatRequestProto.newBuilder()
+ .setDatanodeDetails(datanodeDetails.getProtoBufMessage())
+ .setContainerReport(containerReport)
+ .build();
+ dispatcher.dispatch(heartbeat);
+ verify(containerReportHandler,
+ times(1))
+ .processReport(any(DatanodeDetails.class),
+ any(ContainerReportsProto.class));
+ }
+
+ @Test
+ public void testNodeAndContainerReportDispatcher() throws IOException {
+ Configuration conf = new OzoneConfiguration();
+ SCMDatanodeNodeReportHandler nodeReportHandler =
+ Mockito.mock(SCMDatanodeNodeReportHandler.class);
+ SCMDatanodeContainerReportHandler containerReportHandler =
+ Mockito.mock(SCMDatanodeContainerReportHandler.class);
+ SCMDatanodeHeartbeatDispatcher dispatcher =
+ SCMDatanodeHeartbeatDispatcher.newBuilder(conf, null)
+ .addHandler(NodeReportProto.class, nodeReportHandler)
+ .addHandler(ContainerReportsProto.class, containerReportHandler)
+ .build();
+
+ DatanodeDetails datanodeDetails = TestUtils.getDatanodeDetails();
+ NodeReportProto nodeReport = NodeReportProto.getDefaultInstance();
+ ContainerReportsProto containerReport =
+ ContainerReportsProto.getDefaultInstance();
+ SCMHeartbeatRequestProto heartbeat =
+ SCMHeartbeatRequestProto.newBuilder()
+ .setDatanodeDetails(datanodeDetails.getProtoBufMessage())
+ .setNodeReport(nodeReport)
+ .setContainerReport(containerReport)
+ .build();
+ dispatcher.dispatch(heartbeat);
+ verify(nodeReportHandler,
+ times(1))
+ .processReport(any(DatanodeDetails.class), any(NodeReportProto.class));
+ verify(containerReportHandler,
+ times(1))
+ .processReport(any(DatanodeDetails.class),
+ any(ContainerReportsProto.class));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeNodeReportHandler.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeNodeReportHandler.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeNodeReportHandler.java
new file mode 100644
index 0000000..30a753c
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeNodeReportHandler.java
@@ -0,0 +1,36 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.report;
+
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Test cases to verify TestSCMDatanodeNodeReportHandler's behavior.
+ */
+public class TestSCMDatanodeNodeReportHandler {
+
+
+ //TODO: add test cases to verify SCMDatanodeNodeReportHandler.
+
+ @Test
+ public void dummyTest() {
+ Assert.assertTrue(true);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeReportHandlerFactory.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeReportHandlerFactory.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeReportHandlerFactory.java
new file mode 100644
index 0000000..4b918f7
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/TestSCMDatanodeReportHandlerFactory.java
@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.report;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.NodeReportProto;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Test cases to verify the functionality of SCMDatanodeReportHandlerFactory.
+ */
+public class TestSCMDatanodeReportHandlerFactory {
+
+ @Test
+ public void testNodeReportHandlerConstruction() {
+ Configuration conf = new OzoneConfiguration();
+ SCMDatanodeReportHandlerFactory factory =
+ new SCMDatanodeReportHandlerFactory(conf, null);
+ Assert.assertTrue(factory.getHandlerFor(NodeReportProto.class)
+ instanceof SCMDatanodeNodeReportHandler);
+ }
+
+ @Test
+ public void testContainerReporttHandlerConstruction() {
+ Configuration conf = new OzoneConfiguration();
+ SCMDatanodeReportHandlerFactory factory =
+ new SCMDatanodeReportHandlerFactory(conf, null);
+ Assert.assertTrue(factory.getHandlerFor(ContainerReportsProto.class)
+ instanceof SCMDatanodeContainerReportHandler);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/package-info.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/package-info.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/package-info.java
new file mode 100644
index 0000000..4a3f59f
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/server/report/package-info.java
@@ -0,0 +1,21 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.report;
+/**
+ * Contains test-cases to test Datanode report handlers in SCM.
+ */
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/ddd09d59/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMMetrics.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMMetrics.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMMetrics.java
index 1dbe760..ecddf8e 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMMetrics.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMMetrics.java
@@ -28,6 +28,8 @@ import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
import org.apache.hadoop.hdds.scm.TestUtils;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.server.report
+ .SCMDatanodeContainerReportHandler;
import org.apache.hadoop.metrics2.MetricsRecordBuilder;
import org.apache.hadoop.ozone.MiniOzoneCluster;
import org.apache.hadoop.ozone.OzoneConsts;
@@ -80,7 +82,11 @@ public class TestSCMMetrics {
DatanodeDetails fstDatanodeDetails = TestUtils.getDatanodeDetails();
ContainerReportsProto request = createContainerReport(numReport, stat);
String fstDatanodeUuid = fstDatanodeDetails.getUuidString();
- scmManager.getDatanodeProtocolServer().processContainerReports(
+ SCMDatanodeContainerReportHandler containerReportHandler =
+ new SCMDatanodeContainerReportHandler();
+ containerReportHandler.setConf(conf);
+ containerReportHandler.init(scmManager);
+ containerReportHandler.processReport(
fstDatanodeDetails, request);
// verify container stat metrics
@@ -105,7 +111,7 @@ public class TestSCMMetrics {
DatanodeDetails sndDatanodeDetails = TestUtils.getDatanodeDetails();
request = createContainerReport(1, stat);
String sndDatanodeUuid = sndDatanodeDetails.getUuidString();
- scmManager.getDatanodeProtocolServer().processContainerReports(
+ containerReportHandler.processReport(
sndDatanodeDetails, request);
scmMetrics = getMetrics(SCMMetrics.SOURCE_NAME);
@@ -128,11 +134,11 @@ public class TestSCMMetrics {
// Re-send reports but with different value for validating
// the aggregation.
stat = new ContainerStat(100, 50, 3, 50, 60, 5, 6);
- scmManager.getDatanodeProtocolServer().processContainerReports(
+ containerReportHandler.processReport(
fstDatanodeDetails, createContainerReport(1, stat));
stat = new ContainerStat(1, 1, 1, 1, 1, 1, 1);
- scmManager.getDatanodeProtocolServer().processContainerReports(
+ containerReportHandler.processReport(
sndDatanodeDetails, createContainerReport(1, stat));
// the global container metrics value should be updated
@@ -176,8 +182,12 @@ public class TestSCMMetrics {
DatanodeDetails datanodeDetails = cluster.getHddsDatanodes().get(0)
.getDatanodeDetails();
+ SCMDatanodeContainerReportHandler containerReportHandler =
+ new SCMDatanodeContainerReportHandler();
+ containerReportHandler.setConf(conf);
+ containerReportHandler.init(scmManager);
ContainerReportsProto request = createContainerReport(numReport, stat);
- scmManager.getDatanodeProtocolServer().processContainerReports(
+ containerReportHandler.processReport(
datanodeDetails, request);
MetricsRecordBuilder scmMetrics = getMetrics(SCMMetrics.SOURCE_NAME);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org