[GitHub] [ozone] smengcl commented on a change in pull request #2182: HDDS-5111. DataNode should not always report full information in heartbeat

[GitBox <gi...@apache.org>]

smengcl commented on a change in pull request #2182:
URL: https://github.com/apache/ozone/pull/2182#discussion_r644472508



##########
File path: hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
##########
@@ -248,29 +266,46 @@ public boolean getShutdownOnError() {
    *
    * @param report report to be added
    */
-  public void addReport(GeneratedMessage report) {
+  public void addIncrementalReport(GeneratedMessage report) {
     if (report == null) {
       return;
     }
     final Descriptor descriptor = report.getDescriptorForType();
     Preconditions.checkState(descriptor != null);
     final String reportType = descriptor.getFullName();
     Preconditions.checkState(reportType != null);
-    if (reportType.equals(CONTAINER_REPORTS_PROTO_NAME)) {
-      containerReports.set(report);
-    } else if (reportType.equals(NODE_REPORT_PROTO_NAME)) {
-      nodeReport.set(report);
-    } else if (reportType.equals(PIPELINE_REPORTS_PROTO_NAME)) {
-      pipelineReports.set(report);
-    } else if (ACCEPTED_INCREMENTAL_REPORT_TYPE_SET.contains(reportType)) {
-      synchronized (incrementalReportsQueue) {
-        for (InetSocketAddress endpoint : endpoints) {
-          incrementalReportsQueue.get(endpoint).add(report);
-        }
+    // in some case, we want to add a fullReportType message
+    // as an incremental message.
+    // see XceiverServerRatis#sendPipelineReport
+    synchronized (incrementalReportsQueue) {
+      for (InetSocketAddress endpoint : endpoints) {
+        incrementalReportsQueue.get(endpoint).add(report);
       }
-    } else {
+    }
+  }
+
+  /**
+   * refresh Full report.
+   *
+   * @param report report to be refreshed
+   */
+  public void refreshFullReport(GeneratedMessage report) {
+    if (report == null) {
+      return;
+    }
+    final Descriptor descriptor = report.getDescriptorForType();
+    Preconditions.checkState(descriptor != null);
+    final String reportType = descriptor.getFullName();
+    Preconditions.checkState(reportType != null);
+    if (!fullReportTypeList.contains(reportType)) {
       throw new IllegalArgumentException(
-          "Unidentified report message type: " + reportType);
+          "not full report message type: " + reportType);
+    }
+    type2Reports.get(reportType).set(report);
+    if (null != fullReportSendIndicator){
+      for (Map<String, AtomicBoolean> mp : fullReportSendIndicator.values()){

Review comment:
       ```suggestion
       if (fullReportSendIndicator != null) {
         for (Map<String, AtomicBoolean> mp : fullReportSendIndicator.values()) {
   ```

##########
File path: hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
##########
@@ -116,6 +113,11 @@
   private boolean shutdownOnError = false;
   private boolean shutdownGracefully = false;
   private final AtomicLong threadPoolNotAvailableCount;
+  private Map<InetSocketAddress,
+      Map<String, AtomicBoolean>> fullReportSendIndicator;
+
+  private List<String> fullReportTypeList;
+  private Map<String, AtomicReference<GeneratedMessage>> type2Reports;

Review comment:
       nits
   
   ```suggestion
     // Endpoint -> ReportType -> Boolean of whether the full report should be
     //  queued in getFullReports call.
     private final Map<InetSocketAddress,
         Map<String, AtomicBoolean>> fullReportSendIndicator;
     // List of supported full report types.
     private final List<String> fullReportTypeList;
     // ReportType -> Report.
     private final Map<String, AtomicReference<GeneratedMessage>> type2Reports;
   ```
   
   Pls check the comment, CMIIW

##########
File path: hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
##########
@@ -332,19 +363,21 @@ public void putBackReports(List<GeneratedMessage> reportsToPutBack,
     return reportsToReturn;
   }
 
-  List<GeneratedMessage> getNonIncrementalReports() {
+  List<GeneratedMessage> getNonIncrementalReports(
+      InetSocketAddress endpoint) {
+    Map<String, AtomicBoolean> mp = fullReportSendIndicator.get(endpoint);
     List<GeneratedMessage> nonIncrementalReports = new LinkedList<>();
-    GeneratedMessage report = containerReports.get();
-    if (report != null) {
-      nonIncrementalReports.add(report);
-    }
-    report = nodeReport.get();
-    if (report != null) {
-      nonIncrementalReports.add(report);
-    }
-    report = pipelineReports.get();
-    if (report != null) {
-      nonIncrementalReports.add(report);
+    if (null != mp){
+      for (Map.Entry<String, AtomicBoolean> kv : mp.entrySet()) {
+        if (kv.getValue().get()) {
+          String reportType = kv.getKey();
+          GeneratedMessage msg = type2Reports.get(reportType).get();
+          if (null != msg) {
+            nonIncrementalReports.add(msg);
+            mp.get(reportType).set(false);
+          }

Review comment:
       IIUC `msg` here can never be null as long as the `reportType` is valid because of `initReportTypeCollection`.
   
   Actually we might even want to throw exception if msg is indeed null somehow to catch potential problem earlier in the future.

##########
File path: hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
##########
@@ -788,6 +821,11 @@ public void addEndpoint(InetSocketAddress endpoint) {
       this.containerActions.put(endpoint, new LinkedList<>());
       this.pipelineActions.put(endpoint, new LinkedList<>());
       this.incrementalReportsQueue.put(endpoint, new LinkedList<>());
+      Map<String, AtomicBoolean> mp = new HashMap<>();
+      fullReportTypeList.forEach(e ->{

Review comment:
       ```suggestion
         fullReportTypeList.forEach(e -> {
   ```

##########
File path: hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java
##########
@@ -332,19 +363,21 @@ public void putBackReports(List<GeneratedMessage> reportsToPutBack,
     return reportsToReturn;
   }
 
-  List<GeneratedMessage> getNonIncrementalReports() {
+  List<GeneratedMessage> getNonIncrementalReports(

Review comment:
       Let's rename this to `getFullReports()` then.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org