class ContainerReportHandler extends AbstractContainerReportHandler implements EventHandler<ContainerReportFromDatanode> {

  @Override
  public void onMessage(final ContainerReportFromDatanode reportFromDatanode,
                        final EventPublisher publisher) {

    final DatanodeDetails dnFromReport =
        reportFromDatanode.getDatanodeDetails();
    DatanodeDetails datanodeDetails =
        nodeManager.getNodeByUuid(dnFromReport.getUuidString());
    if (datanodeDetails == null) {
      LOG.warn("Received container report from unknown datanode {}",
          dnFromReport);
      return;
    }
    final ContainerReportsProto containerReport =
        reportFromDatanode.getReport();

    try {
      final List<ContainerReplicaProto> replicas =
          containerReport.getReportsList();
      final Set<ContainerID> containersInSCM =
          nodeManager.getContainers(datanodeDetails);

      final Set<ContainerID> containersInDn = replicas.parallelStream()
          .map(ContainerReplicaProto::getContainerID)
          .map(ContainerID::valueof).collect(Collectors.toSet());

      final Set<ContainerID> missingReplicas = new HashSet<>(containersInSCM);
      missingReplicas.removeAll(containersInDn);

      processContainerReplicas(datanodeDetails, replicas, publisher);
      processMissingReplicas(datanodeDetails, missingReplicas);
      updateDeleteTransaction(datanodeDetails, replicas, publisher);

      /*
       * Update the latest set of containers for this datanode in
       * NodeManager
       */
      nodeManager.setContainers(datanodeDetails, containersInDn);

      containerManager.notifyContainerReportProcessing(true, true);
    } catch (NodeNotFoundException ex) {
      containerManager.notifyContainerReportProcessing(true, false);
      LOG.error("Received container report from unknown datanode {}.",
          datanodeDetails, ex);
    }

  }

  public  ContainerReportHandler(final NodeManager nodeManager,
                                final ContainerManager containerManager,
                                OzoneConfiguration conf);
  public  ContainerReportHandler(final NodeManager nodeManager,
      final ContainerManager containerManager);

}