From 0526bcb6d55e66f17a3600326dd2c8caed472d18 Mon Sep 17 00:00:00 2001 From: sadanand48 Date: Fri, 6 Nov 2026 13:51:10 +0530 Subject: [PATCH 1/8] HDDS-17724. Limit command-status report size in datanode heartbeats. --- .../apache/hadoop/hdds/HddsConfigKeys.java | 4 + .../src/main/resources/ozone-default.xml | 11 + .../statemachine/DatanodeQueueMetrics.java | 129 ++++++++++- .../common/statemachine/StateContext.java | 216 +++++++++++++++++- .../endpoint/HeartbeatEndpointTask.java | 89 +++++++- .../common/statemachine/TestStateContext.java | 131 +++++++++++ .../endpoint/TestHeartbeatEndpointTask.java | 193 +++++++++++++++- 7 files changed, 747 insertions(+), 26 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java index aae7259d813e..ba5c37eae580 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java @@ -73,6 +73,10 @@ public final class HddsConfigKeys { "hdds.heartbeat.report.max.limit"; public static final int HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT = 10000; + public static final String HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE = + "hdds.heartbeat.max.command.status.report.size"; + public static final String + HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE_DEFAULT = "64MB"; // Configuration to allow volume choosing policy. public static final String HDDS_DATANODE_VOLUME_CHOOSING_POLICY = "hdds.datanode.volume.choosing.policy"; diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml index 19b8bce6b5db..ee71cba3d1b4 100644 --- a/hadoop-hdds/common/src/main/resources/ozone-default.xml +++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml @@ -1856,6 +1856,17 @@ 1 are treated as 1. + + hdds.heartbeat.max.command.status.report.size + 64MB + DATANODE + + Maximum serialized size of command status reports included in one + heartbeat. The effective limit can be lower when other heartbeat fields + consume the safe RPC payload budget. Reports beyond the limit stay queued + for a later heartbeat. + + hdds.db.profile DISK diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java index f376d640a3c4..fd5a4a3d7846 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java @@ -23,6 +23,7 @@ import com.google.common.base.CaseFormat; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.text.WordUtils; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; import org.apache.hadoop.hdds.scm.net.HostAndPort; @@ -60,6 +61,16 @@ public final class DatanodeQueueMetrics implements MetricsSource { "ContainerActionQueue"; public static final String PIPELINE_ACTION_QUEUE_PREFIX = "PipelineActionQueue"; + public static final String COMMAND_STATUS_REPORT_QUEUE_PREFIX = + "CommandStatusReportQueue"; + public static final String COMMAND_STATUS_REPORT_SENT_PREFIX = + "CommandStatusReportSent"; + public static final String COMMAND_STATUS_REPORT_DEFERRED_PREFIX = + "CommandStatusReportDeferred"; + public static final String COMMAND_STATUS_REPORT_REQUEUED_PREFIX = + "CommandStatusReportRequeued"; + public static final String COMMAND_STATUS_REPORT_OVERSIZED_PREFIX = + "CommandStatusReportOversized"; private MetricsRegistry registry; @@ -71,6 +82,15 @@ public final class DatanodeQueueMetrics implements MetricsSource { private Map incrementalReportsQueueMap; private Map containerActionQueueMap; private Map pipelineActionQueueMap; + private Map commandStatusReportQueueBytesMap; + private Map commandStatusReportSentBytesMap; + private Map commandStatusReportDeferredBytesMap; + private Map commandStatusReportRequeuedBytesMap; + private Map oversizedCommandStatusReportBytesMap; + private Map commandStatusReportSentBytes; + private Map commandStatusReportDeferredBytes; + private Map commandStatusReportRequeuedBytes; + private Map oversizedCommandStatusReportBytes; public DatanodeQueueMetrics(DatanodeStateMachine datanodeStateMachine) { this.registry = new MetricsRegistry(METRICS_SOURCE_NAME); @@ -110,6 +130,15 @@ private void initializeQueues() { incrementalReportsQueueMap = new HashMap<>(); containerActionQueueMap = new HashMap<>(); pipelineActionQueueMap = new HashMap<>(); + commandStatusReportQueueBytesMap = new HashMap<>(); + commandStatusReportSentBytesMap = new HashMap<>(); + commandStatusReportDeferredBytesMap = new HashMap<>(); + commandStatusReportRequeuedBytesMap = new HashMap<>(); + oversizedCommandStatusReportBytesMap = new HashMap<>(); + commandStatusReportSentBytes = new HashMap<>(); + commandStatusReportDeferredBytes = new HashMap<>(); + commandStatusReportRequeuedBytes = new HashMap<>(); + oversizedCommandStatusReportBytes = new HashMap<>(); } @Override @@ -150,6 +179,33 @@ public void getMetrics(MetricsCollector collector, boolean b) { datanodeStateMachine.getContext().getPipelineActionQueueSize() .getOrDefault(entry.getKey(), 0)); } + Map queuedCommandStatusReportBytes = + datanodeStateMachine.getContext().getCommandStatusReportQueueBytes(); + for (Map.Entry entry: + commandStatusReportQueueBytesMap.entrySet()) { + builder.addGauge(entry.getValue(), + queuedCommandStatusReportBytes.getOrDefault(entry.getKey(), 0L)); + } + for (Map.Entry entry: + commandStatusReportSentBytesMap.entrySet()) { + builder.addCounter(entry.getValue(), + commandStatusReportSentBytes.get(entry.getKey()).get()); + } + for (Map.Entry entry: + commandStatusReportDeferredBytesMap.entrySet()) { + builder.addGauge(entry.getValue(), + commandStatusReportDeferredBytes.get(entry.getKey()).get()); + } + for (Map.Entry entry: + commandStatusReportRequeuedBytesMap.entrySet()) { + builder.addCounter(entry.getValue(), + commandStatusReportRequeuedBytes.get(entry.getKey()).get()); + } + for (Map.Entry entry: + oversizedCommandStatusReportBytesMap.entrySet()) { + builder.addGauge(entry.getValue(), + oversizedCommandStatusReportBytes.get(entry.getKey()).get()); + } } public static synchronized void unRegister() { @@ -158,24 +214,79 @@ public static synchronized void unRegister() { } public void addEndpoint(HostAndPort endpoint) { + String endpointName = CaseFormat.UPPER_UNDERSCORE + .to(CaseFormat.UPPER_CAMEL, endpoint.getHostName()); incrementalReportsQueueMap.computeIfAbsent(endpoint, k -> getMetricsInfo(INCREMENTAL_REPORT_QUEUE_PREFIX, - CaseFormat.UPPER_UNDERSCORE - .to(CaseFormat.UPPER_CAMEL, k.getHostName()))); + endpointName)); containerActionQueueMap.computeIfAbsent(endpoint, k -> getMetricsInfo(CONTAINER_ACTION_QUEUE_PREFIX, - CaseFormat.UPPER_UNDERSCORE - .to(CaseFormat.UPPER_CAMEL, k.getHostName()))); + endpointName)); pipelineActionQueueMap.computeIfAbsent(endpoint, k -> getMetricsInfo(PIPELINE_ACTION_QUEUE_PREFIX, - CaseFormat.UPPER_UNDERSCORE - .to(CaseFormat.UPPER_CAMEL, k.getHostName()))); + endpointName)); + commandStatusReportQueueBytesMap.computeIfAbsent(endpoint, + k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_QUEUE_PREFIX, + endpointName, "Queued command status report bytes for ")); + commandStatusReportSentBytesMap.computeIfAbsent(endpoint, + k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_SENT_PREFIX, + endpointName, "Command status report bytes sent to ")); + commandStatusReportDeferredBytesMap.computeIfAbsent(endpoint, + k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_DEFERRED_PREFIX, + endpointName, "Command status report bytes deferred for ")); + commandStatusReportRequeuedBytesMap.computeIfAbsent(endpoint, + k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_REQUEUED_PREFIX, + endpointName, "Command status report bytes requeued for ")); + oversizedCommandStatusReportBytesMap.computeIfAbsent(endpoint, + k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_OVERSIZED_PREFIX, + endpointName, "Oversized command status report bytes for ")); + commandStatusReportSentBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()); + commandStatusReportDeferredBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()); + commandStatusReportRequeuedBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()); + oversizedCommandStatusReportBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()); } public void removeEndpoint(HostAndPort endpoint) { incrementalReportsQueueMap.remove(endpoint); containerActionQueueMap.remove(endpoint); pipelineActionQueueMap.remove(endpoint); + commandStatusReportQueueBytesMap.remove(endpoint); + commandStatusReportSentBytesMap.remove(endpoint); + commandStatusReportDeferredBytesMap.remove(endpoint); + commandStatusReportRequeuedBytesMap.remove(endpoint); + oversizedCommandStatusReportBytesMap.remove(endpoint); + commandStatusReportSentBytes.remove(endpoint); + commandStatusReportDeferredBytes.remove(endpoint); + commandStatusReportRequeuedBytes.remove(endpoint); + oversizedCommandStatusReportBytes.remove(endpoint); + } + + public void incrCommandStatusReportBytesSent( + HostAndPort endpoint, long bytes) { + commandStatusReportSentBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()).addAndGet(bytes); + } + + public void setCommandStatusReportBytesDeferred( + HostAndPort endpoint, long bytes) { + commandStatusReportDeferredBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()).set(bytes); + } + + public void incrCommandStatusReportBytesRequeued( + HostAndPort endpoint, long bytes) { + commandStatusReportRequeuedBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()).addAndGet(bytes); + } + + public void setOversizedCommandStatusReportBytes( + HostAndPort endpoint, long bytes) { + oversizedCommandStatusReportBytes.computeIfAbsent(endpoint, + k -> new AtomicLong()).set(bytes); } @VisibleForTesting @@ -198,4 +309,10 @@ private MetricsInfo getMetricsInfo(String prefix, String metricName) { String description = "Queue size of " + metricName + " from " + prefix; return info(metric, description); } + + private MetricsInfo getByteMetricsInfo(String prefix, String metricName, + String description) { + String metric = prefix + WordUtils.capitalize(metricName) + "Bytes"; + return info(metric, description + metricName); + } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java index ae5352169eb9..49c216dcdfbb 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java @@ -25,6 +25,7 @@ import static org.apache.hadoop.hdds.utils.HddsServerUtil.getScmInitialHeartbeatInterval; import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.CodedOutputStream; import com.google.protobuf.Descriptors.Descriptor; import com.google.protobuf.Message; import java.io.IOException; @@ -36,6 +37,7 @@ import java.util.LinkedHashMap; import java.util.LinkedList; import java.util.List; +import java.util.ListIterator; import java.util.Map; import java.util.Objects; import java.util.OptionalLong; @@ -59,6 +61,7 @@ import org.apache.commons.collections4.CollectionUtils; import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatus.Status; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatusReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; @@ -69,6 +72,7 @@ import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReport; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto; import org.apache.hadoop.hdds.scm.net.HostAndPort; import org.apache.hadoop.hdfs.util.EnumCounters; import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.ClosePipelineCommandHandler; @@ -381,9 +385,8 @@ public List getAllAvailableReports( /** * Returns up to {@code maxLimit} available reports for the endpoint, * removing them from the queue. Any reports beyond the limit stay queued - * and are returned by a later call; callers can use {@link - * #hasPendingReports(HostAndPort)} to detect that case and schedule a - * follow-up heartbeat. + * and are returned by a later call; callers can use + * {@link #hasPendingReports(HostAndPort)} to detect that case. * * @return List of reports */ @@ -392,6 +395,71 @@ public List getAllAvailableReports( return getAllAvailableReportsUpToLimit(endpoint, maxLimit); } + /** + * Returns available reports other than command status reports. Full reports + * are selected before incremental reports, and the total number returned is + * capped by {@code maxLimit}. + */ + public List getAvailableNonCommandStatusReports( + HostAndPort endpoint, int maxLimit) { + List reports = getFullReports(endpoint, maxLimit); + reports.addAll(getIncrementalReportsExcludingCommandStatus( + endpoint, maxLimit - reports.size())); + return reports; + } + + /** + * Selects command status reports within both the report count and serialized + * byte limits. An aggregate report may be split between complete command + * statuses, but an individual command status is never split. + */ + public CommandStatusReportSelection getCommandStatusReports( + HostAndPort endpoint, int maxLimit, long maxBytes, + long maxConfiguredBytes) { + List selected = new LinkedList<>(); + long selectedBytes = 0; + long oversizedBytes = 0; + synchronized (incrementalReportsQueue) { + List reportsForEndpoint = + incrementalReportsQueue.get(endpoint); + if (reportsForEndpoint != null) { + ListIterator iterator = reportsForEndpoint.listIterator(); + while (iterator.hasNext() && selected.size() < maxLimit) { + Message message = iterator.next(); + if (!(message instanceof CommandStatusReportsProto)) { + continue; + } + + CommandStatusReportsProto report = + (CommandStatusReportsProto) message; + long remainingBytes = maxBytes - selectedBytes; + long reportBytes = getCommandStatusReportWireSize(report); + if (reportBytes <= remainingBytes) { + selected.add(report); + selectedBytes += reportBytes; + iterator.remove(); + continue; + } + + CommandStatusReportSplit split = + splitCommandStatusReport(report, remainingBytes); + if (split.getSelected() != null) { + selected.add(split.getSelected()); + selectedBytes += + getCommandStatusReportWireSize(split.getSelected()); + iterator.set(split.getRemainder()); + } else if (getFirstCommandStatusWireSize(report) + > maxConfiguredBytes) { + oversizedBytes = getFirstCommandStatusWireSize(report); + } + break; + } + } + return new CommandStatusReportSelection(selected, selectedBytes, + getCommandStatusReportQueueBytes(endpoint), oversizedBytes); + } + } + /** * Returns true if any reports are still queued for the endpoint. */ @@ -466,6 +534,148 @@ List getIncrementalReports( return reportsToReturn; } + private List getIncrementalReportsExcludingCommandStatus( + HostAndPort endpoint, int maxLimit) { + List reportsToReturn = new LinkedList<>(); + synchronized (incrementalReportsQueue) { + List reportsForEndpoint = + incrementalReportsQueue.get(endpoint); + if (reportsForEndpoint != null) { + Iterator iterator = reportsForEndpoint.iterator(); + while (iterator.hasNext() && reportsToReturn.size() < maxLimit) { + Message report = iterator.next(); + if (!(report instanceof CommandStatusReportsProto)) { + reportsToReturn.add(report); + iterator.remove(); + } + } + } + } + return reportsToReturn; + } + + private CommandStatusReportSplit splitCommandStatusReport( + CommandStatusReportsProto report, long maxBytes) { + CommandStatusReportsProto.Builder selected = + CommandStatusReportsProto.newBuilder(); + CommandStatusReportsProto.Builder remainder = + CommandStatusReportsProto.newBuilder(); + long selectedReportBytes = 0; + boolean limitReached = false; + for (StorageContainerDatanodeProtocolProtos.CommandStatus status : + report.getCmdStatusList()) { + int statusBytes = CodedOutputStream.computeMessageSize( + CommandStatusReportsProto.CMDSTATUS_FIELD_NUMBER, status); + long candidateReportBytes = selectedReportBytes + statusBytes; + if (!limitReached && + getCommandStatusReportWireSize(candidateReportBytes) <= maxBytes) { + selected.addCmdStatus(status); + selectedReportBytes = candidateReportBytes; + } else { + limitReached = true; + remainder.addCmdStatus(status); + } + } + return new CommandStatusReportSplit( + selected.getCmdStatusCount() == 0 ? null : selected.build(), + remainder.build()); + } + + private long getFirstCommandStatusWireSize( + CommandStatusReportsProto report) { + if (report.getCmdStatusCount() == 0) { + return getCommandStatusReportWireSize(report); + } + int statusBytes = CodedOutputStream.computeMessageSize( + CommandStatusReportsProto.CMDSTATUS_FIELD_NUMBER, + report.getCmdStatus(0)); + return getCommandStatusReportWireSize(statusBytes); + } + + private static long getCommandStatusReportWireSize( + CommandStatusReportsProto report) { + return getCommandStatusReportWireSize(report.getSerializedSize()); + } + + private static long getCommandStatusReportWireSize( + long reportBytes) { + return CodedOutputStream.computeTagSize( + SCMHeartbeatRequestProto.COMMANDSTATUSREPORTS_FIELD_NUMBER) + + CodedOutputStream.computeUInt64SizeNoTag(reportBytes) + + reportBytes; + } + + public long getCommandStatusReportQueueBytes(HostAndPort endpoint) { + synchronized (incrementalReportsQueue) { + List reports = incrementalReportsQueue.get(endpoint); + return reports == null ? 0 : reports.stream() + .filter(CommandStatusReportsProto.class::isInstance) + .map(CommandStatusReportsProto.class::cast) + .mapToLong(StateContext::getCommandStatusReportWireSize) + .sum(); + } + } + + public Map getCommandStatusReportQueueBytes() { + synchronized (incrementalReportsQueue) { + return incrementalReportsQueue.keySet().stream() + .collect(Collectors.toMap( + endpoint -> endpoint, + this::getCommandStatusReportQueueBytes)); + } + } + + public static final class CommandStatusReportSelection { + private final List reports; + private final long selectedBytes; + private final long deferredBytes; + private final long oversizedBytes; + + private CommandStatusReportSelection( + List reports, long selectedBytes, + long deferredBytes, long oversizedBytes) { + this.reports = reports; + this.selectedBytes = selectedBytes; + this.deferredBytes = deferredBytes; + this.oversizedBytes = oversizedBytes; + } + + public List getReports() { + return reports; + } + + public long getSelectedBytes() { + return selectedBytes; + } + + public long getDeferredBytes() { + return deferredBytes; + } + + public long getOversizedBytes() { + return oversizedBytes; + } + } + + private static final class CommandStatusReportSplit { + private final CommandStatusReportsProto selected; + private final CommandStatusReportsProto remainder; + + private CommandStatusReportSplit(CommandStatusReportsProto selected, + CommandStatusReportsProto remainder) { + this.selected = selected; + this.remainder = remainder; + } + + private CommandStatusReportsProto getSelected() { + return selected; + } + + private CommandStatusReportsProto getRemainder() { + return remainder; + } + } + List getFullReports( HostAndPort endpoint, int maxLimit) { int count = 0; diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java index 76358e4755c8..f32ce41fa72a 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java @@ -21,10 +21,13 @@ import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_ACTION_MAX_LIMIT_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_MISSED_COUNT_THRESHOLD; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_MISSED_COUNT_THRESHOLD_DEFAULT; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT_DEFAULT; +import static org.apache.hadoop.hdds.conf.StorageUnit.BYTES; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_DEFAULT; import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_CLIENT_FAILOVER_RESOLVE_NEEDED_KEY; import static org.apache.hadoop.ozone.container.upgrade.UpgradeUtils.toVersionProto; @@ -37,11 +40,13 @@ import java.util.LinkedList; import java.util.List; import java.util.concurrent.Callable; +import org.apache.hadoop.fs.CommonConfigurationKeys; import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatusReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerActionsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.DatanodeVersionProto; @@ -57,6 +62,7 @@ import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine; import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine.EndPointStates; import org.apache.hadoop.ozone.container.common.statemachine.StateContext; +import org.apache.hadoop.ozone.container.common.statemachine.StateContext.CommandStatusReportSelection; import org.apache.hadoop.ozone.container.upgrade.DatanodeVersionManager; import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand; import org.apache.hadoop.ozone.protocol.commands.ClosePipelineCommand; @@ -80,6 +86,7 @@ public class HeartbeatEndpointTask implements Callable { private static final Logger LOG = LoggerFactory.getLogger(HeartbeatEndpointTask.class); + private static final double IPC_SIZE_SAFETY_FACTOR = 0.9; private final EndpointStateMachine rpcEndpoint; private final DatanodeDetails datanodeDetails; private StateContext context; @@ -91,6 +98,8 @@ public class HeartbeatEndpointTask private final HDDSVersion testCurrentVersion; private final boolean resolveOnFailureEnabled; private final int refreshThreshold; + private final long maxCommandStatusReportBytes; + private final long safeHeartbeatSize; /** * Constructs a SCM heart beat. @@ -109,6 +118,18 @@ public HeartbeatEndpointTask(EndpointStateMachine rpcEndpoint, HDDS_PIPELINE_ACTION_MAX_LIMIT_DEFAULT); this.maxReportsPerHB = Math.max(1, conf.getInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT)); + this.maxCommandStatusReportBytes = (long) conf.getStorageSize( + HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE, + HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE_DEFAULT, BYTES); + Preconditions.checkArgument(maxCommandStatusReportBytes > 0, + HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE + " must be positive"); + int ipcMaximumDataLength = conf.getInt( + CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, + CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH_DEFAULT); + Preconditions.checkArgument(ipcMaximumDataLength > 0, + CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH + " must be positive"); + this.safeHeartbeatSize = + (long) Math.floor(ipcMaximumDataLength * IPC_SIZE_SAFETY_FACTOR); this.datanodeDetails = context.getParent().getDatanodeDetails(); this.versionManager = context.getParent().getVersionManager(); @@ -136,6 +157,7 @@ public HeartbeatEndpointTask(EndpointStateMachine rpcEndpoint, public EndpointStateMachine.EndPointStates call() throws Exception { rpcEndpoint.lock(); SCMHeartbeatRequestProto.Builder requestBuilder = null; + CommandStatusReportSelection commandStatusSelection = null; try { DatanodeVersionProto versionInfo = toVersionProto( versionManager.getApparentVersion(), @@ -148,10 +170,12 @@ public EndpointStateMachine.EndPointStates call() throws Exception { requestBuilder = SCMHeartbeatRequestProto.newBuilder() .setDatanodeDetails(datanodeDetailsProto) .setDatanodeVersion(versionInfo); - addReports(requestBuilder); + int reportCount = addNonCommandStatusReports(requestBuilder); addContainerActions(requestBuilder); addPipelineActions(requestBuilder); addQueuedCommandCounts(requestBuilder); + commandStatusSelection = addCommandStatusReports( + requestBuilder, maxReportsPerHB - reportCount); SCMHeartbeatRequestProto request = requestBuilder.build(); LOG.debug("Sending heartbeat message : {}", request); SCMHeartbeatResponseProto response = rpcEndpoint.getEndPoint() @@ -159,14 +183,21 @@ public EndpointStateMachine.EndPointStates call() throws Exception { processResponse(response, datanodeDetailsProto); rpcEndpoint.setLastSuccessfulHeartbeat(ZonedDateTime.now()); rpcEndpoint.zeroMissedCount(); - if (context.hasPendingReports(rpcEndpoint.getAddress())) { - // addReports was capped at maxReportsPerHB and left reports queued. Trigger immediately. - context.getParent().setNextHB(Time.monotonicNow()); + if (context.getQueueMetrics() != null) { + context.getQueueMetrics().incrCommandStatusReportBytesSent( + rpcEndpoint.getAddress(), + commandStatusSelection.getSelectedBytes()); } } catch (IOException ex) { Preconditions.checkState(requestBuilder != null); // put back the reports which failed to be sent putBackIncrementalReports(requestBuilder); + if (commandStatusSelection != null && + context.getQueueMetrics() != null) { + context.getQueueMetrics().incrCommandStatusReportBytesRequeued( + rpcEndpoint.getAddress(), + commandStatusSelection.getSelectedBytes()); + } rpcEndpoint.logIfNeeded(ex); maybeRefreshScmAddress(ex); } finally { @@ -211,14 +242,13 @@ private void putBackIncrementalReports( context.putBackReports(reports, rpcEndpoint.getAddress()); } - /** - * Adds all the available reports to heartbeat. - * - * @param requestBuilder builder to which the report has to be added. - */ - private void addReports(SCMHeartbeatRequestProto.Builder requestBuilder) { + private int addNonCommandStatusReports( + SCMHeartbeatRequestProto.Builder requestBuilder) { + int reportCount = 0; for (Message report : - context.getAllAvailableReports(rpcEndpoint.getAddress(), maxReportsPerHB)) { + context.getAvailableNonCommandStatusReports( + rpcEndpoint.getAddress(), maxReportsPerHB)) { + reportCount++; String reportName = report.getDescriptorForType().getFullName(); for (Descriptors.FieldDescriptor descriptor : SCMHeartbeatRequestProto.getDescriptor().getFields()) { @@ -233,6 +263,43 @@ private void addReports(SCMHeartbeatRequestProto.Builder requestBuilder) { } } } + return reportCount; + } + + private CommandStatusReportSelection addCommandStatusReports( + SCMHeartbeatRequestProto.Builder requestBuilder, + int remainingReportCount) { + long currentHeartbeatSize = requestBuilder.build().getSerializedSize(); + long remainingSafeBytes = Math.max( + 0, safeHeartbeatSize - currentHeartbeatSize); + long commandStatusReportBudget = Math.min( + maxCommandStatusReportBytes, remainingSafeBytes); + CommandStatusReportSelection selection = + context.getCommandStatusReports(rpcEndpoint.getAddress(), + remainingReportCount, commandStatusReportBudget, + maxCommandStatusReportBytes); + for (CommandStatusReportsProto report : selection.getReports()) { + requestBuilder.addCommandStatusReports(report); + } + + if (currentHeartbeatSize > safeHeartbeatSize) { + LOG.warn("Heartbeat for {} is already {} bytes before command status " + + "reports, exceeding the safe size of {} bytes", + rpcEndpoint.getAddress(), currentHeartbeatSize, safeHeartbeatSize); + } + if (selection.getOversizedBytes() > 0) { + LOG.warn("A command status for {} is {} bytes, exceeding the configured " + + "per-heartbeat command status report limit of {} bytes", + rpcEndpoint.getAddress(), selection.getOversizedBytes(), + maxCommandStatusReportBytes); + } + if (context.getQueueMetrics() != null) { + context.getQueueMetrics().setCommandStatusReportBytesDeferred( + rpcEndpoint.getAddress(), selection.getDeferredBytes()); + context.getQueueMetrics().setOversizedCommandStatusReportBytes( + rpcEndpoint.getAddress(), selection.getOversizedBytes()); + } + return selection; } /** diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java index 249b0ac788cf..852cefad118a 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java @@ -30,6 +30,7 @@ import static org.mockito.Mockito.when; import com.google.common.util.concurrent.ThreadFactoryBuilder; +import com.google.protobuf.CodedOutputStream; import com.google.protobuf.Descriptors.Descriptor; import com.google.protobuf.Message; import java.io.IOException; @@ -52,18 +53,22 @@ import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatus; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatusReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.IncrementalContainerReportProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineAction; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReport; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMHeartbeatRequestProto; import org.apache.hadoop.hdds.scm.net.HostAndPort; import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.hdfs.util.EnumCounters; import org.apache.hadoop.ozone.container.common.ContainerTestUtils; import org.apache.hadoop.ozone.container.common.impl.ContainerSet; import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine.DatanodeStates; +import org.apache.hadoop.ozone.container.common.statemachine.StateContext.CommandStatusReportSelection; import org.apache.hadoop.ozone.container.common.states.DatanodeState; import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; import org.apache.hadoop.ozone.protocol.commands.CloseContainerCommand; @@ -248,6 +253,109 @@ public void testGetAllAvailableReportsRespectsLimit() throws IOException { assertEquals(0, ctx.getAllAvailableReports(scm1, 4).size()); } + @Test + public void testCommandStatusReportsRespectByteLimit() throws IOException { + StateContext context = createSubject(); + HostAndPort scm = new HostAndPort("scm", 9001); + context.addEndpoint(scm); + + CommandStatusReportsProto first = newCommandStatusReport(1, 128); + CommandStatusReportsProto second = newCommandStatusReport(2, 128); + context.addIncrementalReport(first); + context.addIncrementalReport(second); + + long firstBytes = commandStatusReportWireSize(first); + CommandStatusReportSelection selection = context.getCommandStatusReports( + scm, 10, firstBytes, firstBytes); + assertEquals(Collections.singletonList(first), selection.getReports()); + assertEquals(firstBytes, selection.getSelectedBytes()); + assertEquals(commandStatusReportWireSize(second), + selection.getDeferredBytes()); + } + + @Test + public void testAggregateCommandStatusReportIsSplitAtStatusBoundary() + throws IOException { + StateContext context = createSubject(); + HostAndPort scm = new HostAndPort("scm", 9001); + context.addEndpoint(scm); + + CommandStatus first = newCommandStatus(1, 128); + CommandStatus second = newCommandStatus(2, 128); + CommandStatus third = newCommandStatus(3, 128); + CommandStatusReportsProto aggregate = + CommandStatusReportsProto.newBuilder() + .addCmdStatus(first) + .addCmdStatus(second) + .addCmdStatus(third) + .build(); + CommandStatusReportsProto firstBatch = + CommandStatusReportsProto.newBuilder() + .addCmdStatus(first) + .addCmdStatus(second) + .build(); + CommandStatusReportsProto secondBatch = + CommandStatusReportsProto.newBuilder() + .addCmdStatus(third) + .build(); + context.addIncrementalReport(aggregate); + + CommandStatusReportSelection selection = context.getCommandStatusReports( + scm, 10, commandStatusReportWireSize(firstBatch), + commandStatusReportWireSize(firstBatch)); + assertEquals(Collections.singletonList(firstBatch), + selection.getReports()); + assertEquals(commandStatusReportWireSize(secondBatch), + selection.getDeferredBytes()); + + selection = context.getCommandStatusReports( + scm, 10, commandStatusReportWireSize(firstBatch), + commandStatusReportWireSize(firstBatch)); + assertEquals(Collections.singletonList(secondBatch), + selection.getReports()); + assertEquals(0, selection.getDeferredBytes()); + } + + @Test + public void testNonCommandReportsAreSelectedIndependently() + throws IOException { + StateContext context = createSubject(); + HostAndPort scm = new HostAndPort("scm", 9001); + context.addEndpoint(scm); + CommandStatusReportsProto commandStatus = + newCommandStatusReport(1, 128); + IncrementalContainerReportProto icr = + IncrementalContainerReportProto.getDefaultInstance(); + context.addIncrementalReport(commandStatus); + context.addIncrementalReport(icr); + + assertEquals(Collections.singletonList(icr), + context.getAvailableNonCommandStatusReports(scm, 10)); + assertEquals(commandStatusReportWireSize(commandStatus), + context.getCommandStatusReportQueueBytes(scm)); + } + + @Test + public void testOversizedCommandStatusRemainsQueued() throws IOException { + StateContext context = createSubject(); + HostAndPort scm = new HostAndPort("scm", 9001); + context.addEndpoint(scm); + CommandStatusReportsProto report = newCommandStatusReport(1, 128); + context.addIncrementalReport(report); + long reportBytes = commandStatusReportWireSize(report); + + CommandStatusReportSelection selection = context.getCommandStatusReports( + scm, 10, reportBytes - 1, reportBytes - 1); + assertTrue(selection.getReports().isEmpty()); + assertEquals(reportBytes, selection.getDeferredBytes()); + assertEquals(reportBytes, selection.getOversizedBytes()); + + selection = context.getCommandStatusReports( + scm, 10, reportBytes, reportBytes); + assertEquals(Collections.singletonList(report), selection.getReports()); + assertEquals(0, selection.getDeferredBytes()); + } + @Test public void testHasPendingReportsWithFullReports() throws IOException { StateContext ctx = createSubject(); @@ -294,6 +402,29 @@ void checkReportCount(List reports, assertEquals(expectedReportCount, reportCount); } + private static CommandStatusReportsProto newCommandStatusReport( + long commandId, int messageSize) { + return CommandStatusReportsProto.newBuilder() + .addCmdStatus(newCommandStatus(commandId, messageSize)) + .build(); + } + + private static CommandStatus newCommandStatus( + long commandId, int messageSize) { + return CommandStatus.newBuilder() + .setCmdId(commandId) + .setStatus(CommandStatus.Status.EXECUTED) + .setType(SCMCommandProto.Type.deleteBlocksCommand) + .setMsg(String.join("", Collections.nCopies(messageSize, "x"))) + .build(); + } + + private static long commandStatusReportWireSize( + CommandStatusReportsProto report) { + return CodedOutputStream.computeMessageSize( + SCMHeartbeatRequestProto.COMMANDSTATUSREPORTS_FIELD_NUMBER, report); + } + /** * Check if Container, Node and Pipeline report APIs work as expected. */ diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java index 4f90b767cdcd..892c25304d19 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java @@ -18,6 +18,7 @@ package org.apache.hadoop.ozone.container.common.states.endpoint; import static java.util.Collections.emptyList; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT; import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type.reconcileContainerCommand; import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type.reconstructECContainersCommand; @@ -28,17 +29,22 @@ import static org.mockito.Mockito.any; import static org.mockito.Mockito.anyLong; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import com.google.protobuf.CodedOutputStream; import com.google.protobuf.UnsafeByteOperations; +import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.OptionalLong; import java.util.Set; import java.util.UUID; +import org.apache.hadoop.fs.CommonConfigurationKeys; import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.conf.ConfigurationSource; @@ -46,8 +52,10 @@ import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatus; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatusReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.IncrementalContainerReportProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto; @@ -62,6 +70,7 @@ import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine.DatanodeStates; import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine; import org.apache.hadoop.ozone.container.common.statemachine.StateContext; +import org.apache.hadoop.ozone.container.common.statemachine.StateContext.CommandStatusReportSelection; import org.apache.hadoop.ozone.container.upgrade.DatanodeVersionManager; import org.apache.hadoop.ozone.protocol.commands.ReconcileContainerCommand; import org.apache.hadoop.ozone.protocol.commands.ReconstructECContainersCommand; @@ -405,7 +414,8 @@ public void testheartbeatWithAllReports() throws Exception { } @Test - public void testHeartbeatCapsReportsAndSchedulesFollowup() throws Exception { + public void testHeartbeatCapsReportsWithoutImmediateFollowup() + throws Exception { OzoneConfiguration conf = new OzoneConfiguration(); conf.setInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, 2); DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); @@ -441,16 +451,169 @@ public void testHeartbeatCapsReportsAndSchedulesFollowup() throws Exception { // Only the capped number of ICRs is sent in this heartbeat. SCMHeartbeatRequestProto heartbeat = argument.getValue(); assertEquals(2, heartbeat.getIncrementalContainerReportCount()); - // The remainder stays queued and a follow-up heartbeat is scheduled now. + // The remainder stays queued for the next regular heartbeat. assertTrue(context.hasPendingReports(TEST_SCM_ENDPOINT)); - verify(datanodeStateMachine).setNextHB(anyLong()); + verify(datanodeStateMachine, never()).setNextHB(anyLong()); - // The follow-ups drain the rest; the last one sends 1 ICR and schedules nothing. + // Subsequent regular heartbeats drain the rest. endpointTask.call(); endpointTask.call(); assertEquals(1, argument.getValue().getIncrementalContainerReportCount()); assertFalse(context.hasPendingReports(TEST_SCM_ENDPOINT)); - verify(datanodeStateMachine, times(2)).setNextHB(anyLong()); + verify(datanodeStateMachine, never()).setNextHB(anyLong()); + } + + @Test + public void testHeartbeatCapsCommandStatusReportsByBytes() + throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, 10); + CommandStatusReportsProto first = newCommandStatusReport(1, 128); + CommandStatusReportsProto second = newCommandStatusReport(2, 128); + long reportBytes = commandStatusReportWireSize(first); + conf.set(HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE, + reportBytes + "B"); + DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, + datanodeStateMachine, ""); + when(datanodeStateMachine.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + + StorageContainerDatanodeProtocolClientSideTranslatorPB scm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + ArgumentCaptor argument = ArgumentCaptor + .forClass(SCMHeartbeatRequestProto.class); + when(scm.sendHeartbeat(argument.capture())) + .thenAnswer(invocation -> + SCMHeartbeatResponseProto.newBuilder() + .setDatanodeUUID( + ((SCMHeartbeatRequestProto) invocation.getArgument(0)) + .getDatanodeDetails().getUuid()) + .build()); + + HeartbeatEndpointTask endpointTask = getHeartbeatEndpointTask( + conf, context, scm); + context.addEndpoint(TEST_SCM_ENDPOINT); + context.addIncrementalReport(first); + context.addIncrementalReport(second); + + endpointTask.call(); + + assertEquals(1, argument.getValue().getCommandStatusReportsCount()); + assertEquals(first, + argument.getValue().getCommandStatusReports(0)); + assertTrue(context.hasPendingReports(TEST_SCM_ENDPOINT)); + verify(datanodeStateMachine, never()).setNextHB(anyLong()); + } + + @Test + public void testExistingHeartbeatContentReducesCommandStatusBudget() + throws Exception { + IncrementalContainerReportProto icr = + IncrementalContainerReportProto.newBuilder() + .addReport(ContainerReplicaProto.newBuilder() + .setContainerID(1) + .setState(ContainerReplicaProto.State.CLOSED) + .setOriginNodeId(String.join("", Collections.nCopies( + 1024, "x"))) + .build()) + .build(); + DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + when(datanodeStateMachine.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + + OzoneConfiguration baselineConf = new OzoneConfiguration(); + StateContext baselineContext = new StateContext(baselineConf, + DatanodeStates.RUNNING, datanodeStateMachine, ""); + StorageContainerDatanodeProtocolClientSideTranslatorPB baselineScm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + ArgumentCaptor baselineArgument = + ArgumentCaptor.forClass(SCMHeartbeatRequestProto.class); + when(baselineScm.sendHeartbeat(baselineArgument.capture())) + .thenAnswer(invocation -> + SCMHeartbeatResponseProto.newBuilder() + .setDatanodeUUID( + ((SCMHeartbeatRequestProto) invocation.getArgument(0)) + .getDatanodeDetails().getUuid()) + .build()); + HeartbeatEndpointTask baselineTask = getHeartbeatEndpointTask( + baselineConf, baselineContext, baselineScm); + baselineContext.addEndpoint(TEST_SCM_ENDPOINT); + baselineContext.addIncrementalReport(icr); + baselineTask.call(); + long nonCommandHeartbeatBytes = + baselineArgument.getValue().getSerializedSize(); + + CommandStatusReportsProto first = newCommandStatusReport(1, 128); + CommandStatusReportsProto second = newCommandStatusReport(2, 128); + long reportBytes = commandStatusReportWireSize(first); + long targetSafeSize = nonCommandHeartbeatBytes + reportBytes; + int ipcMaximumDataLength = + (int) Math.ceil(targetSafeSize / 0.9D); + long safeHeartbeatSize = + (long) Math.floor(ipcMaximumDataLength * 0.9D); + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, + ipcMaximumDataLength); + conf.set(HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE, + (reportBytes * 2) + "B"); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, + datanodeStateMachine, ""); + StorageContainerDatanodeProtocolClientSideTranslatorPB scm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + ArgumentCaptor argument = + ArgumentCaptor.forClass(SCMHeartbeatRequestProto.class); + when(scm.sendHeartbeat(argument.capture())) + .thenAnswer(invocation -> + SCMHeartbeatResponseProto.newBuilder() + .setDatanodeUUID( + ((SCMHeartbeatRequestProto) invocation.getArgument(0)) + .getDatanodeDetails().getUuid()) + .build()); + HeartbeatEndpointTask endpointTask = getHeartbeatEndpointTask( + conf, context, scm); + context.addEndpoint(TEST_SCM_ENDPOINT); + context.addIncrementalReport(icr); + context.addIncrementalReport(first); + context.addIncrementalReport(second); + + endpointTask.call(); + + assertEquals(1, argument.getValue().getCommandStatusReportsCount()); + assertTrue(argument.getValue().getSerializedSize() <= safeHeartbeatSize); + assertTrue(context.hasPendingReports(TEST_SCM_ENDPOINT)); + } + + @Test + public void testFailedHeartbeatRestoresCommandStatusReports() + throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + CommandStatusReportsProto first = newCommandStatusReport(1, 128); + CommandStatusReportsProto second = newCommandStatusReport(2, 128); + long reportBytes = commandStatusReportWireSize(first); + conf.set(HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE, + reportBytes + "B"); + DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, + datanodeStateMachine, ""); + when(datanodeStateMachine.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + + StorageContainerDatanodeProtocolClientSideTranslatorPB scm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + when(scm.sendHeartbeat(any())).thenThrow(new IOException("failure")); + + HeartbeatEndpointTask endpointTask = getHeartbeatEndpointTask( + conf, context, scm); + context.addEndpoint(TEST_SCM_ENDPOINT); + context.addIncrementalReport(first); + context.addIncrementalReport(second); + + endpointTask.call(); + + CommandStatusReportSelection selection = context.getCommandStatusReports( + TEST_SCM_ENDPOINT, 10, reportBytes * 2, reportBytes * 2); + assertEquals(Arrays.asList(first, second), selection.getReports()); } @ParameterizedTest @@ -626,4 +789,22 @@ private ContainerAction getContainerAction() { .setReason(ContainerAction.Reason.CONTAINER_FULL); return builder.build(); } + + private static CommandStatusReportsProto newCommandStatusReport( + long commandId, int messageSize) { + return CommandStatusReportsProto.newBuilder() + .addCmdStatus(CommandStatus.newBuilder() + .setCmdId(commandId) + .setStatus(CommandStatus.Status.EXECUTED) + .setType(SCMCommandProto.Type.deleteBlocksCommand) + .setMsg(String.join("", Collections.nCopies(messageSize, "x"))) + .build()) + .build(); + } + + private static long commandStatusReportWireSize( + CommandStatusReportsProto report) { + return CodedOutputStream.computeMessageSize( + SCMHeartbeatRequestProto.COMMANDSTATUSREPORTS_FIELD_NUMBER, report); + } } From 953d4bf182d5e16ef85e46a52b5362ce086000c5 Mon Sep 17 00:00:00 2001 From: sadanand48 Date: Fri, 6 Nov 2026 19:31:30 +0530 Subject: [PATCH 2/8] fix: address review comment on hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java --- .../common/states/endpoint/TestHeartbeatEndpointTask.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java index 892c25304d19..2c6417fcc0ed 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java @@ -414,7 +414,7 @@ public void testheartbeatWithAllReports() throws Exception { } @Test - public void testHeartbeatCapsReportsWithoutImmediateFollowup() + public void testHeartbeatCapsReports() throws Exception { OzoneConfiguration conf = new OzoneConfiguration(); conf.setInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, 2); From acf68b81f6faa19e4d28d3975eed04e45d61dad5 Mon Sep 17 00:00:00 2001 From: sadanand48 Date: Mon, 9 Nov 2026 12:15:50 +0530 Subject: [PATCH 3/8] address comments --- .../apache/hadoop/hdds/HddsConfigKeys.java | 7 +- .../org/apache/hadoop/hdds/scm/ScmConfig.java | 7 +- .../src/main/resources/ozone-default.xml | 11 +- .../statemachine/DatanodeQueueMetrics.java | 42 +++---- .../common/statemachine/StateContext.java | 90 +++++--------- .../endpoint/HeartbeatEndpointTask.java | 27 ++--- .../common/statemachine/TestStateContext.java | 89 ++++---------- .../endpoint/TestHeartbeatEndpointTask.java | 114 ++++++++++++++---- 8 files changed, 194 insertions(+), 193 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java index ba5c37eae580..5e44f1ee19c0 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java @@ -73,10 +73,9 @@ public final class HddsConfigKeys { "hdds.heartbeat.report.max.limit"; public static final int HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT = 10000; - public static final String HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE = - "hdds.heartbeat.max.command.status.report.size"; - public static final String - HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE_DEFAULT = "64MB"; + public static final String HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE = + "hdds.heartbeat.command.status.report.max.size"; + public static final String HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE_DEFAULT = "64MB"; // Configuration to allow volume choosing policy. public static final String HDDS_DATANODE_VOLUME_CHOOSING_POLICY = "hdds.datanode.volume.choosing.policy"; diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java index 9d3d7ef85cbe..0ddf71f9eeee 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java @@ -112,7 +112,12 @@ public class ScmConfig extends ReconfigurableConfig { + "The block num is counted at the replica level." + "If SCM has 100000 blocks which need to be deleted and the " + "configuration is 5000 then it would only send 5000 blocks " - + "for deletion to the datanodes." + + "for deletion to the datanodes. Increasing this value can " + + "increase delete command acknowledgement size; each " + + "acknowledgement must fit within the datanode heartbeat " + + "command status report limit " + + "(hdds.heartbeat.command.status.report.max.size) and the IPC " + + "size limit (ipc.maximum.data.length)." ) private int blockDeletionLimit; diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml index ee71cba3d1b4..606592652018 100644 --- a/hadoop-hdds/common/src/main/resources/ozone-default.xml +++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml @@ -1852,19 +1852,22 @@ Maximum number of report messages (not containers) a datanode includes in a single heartbeat to an SCM or Recon endpoint. Reports beyond the limit - stay queued and are sent in an immediate follow-up heartbeat. Values below - 1 are treated as 1. + stay queued and are sent in later scheduled heartbeats. Values below 1 + are treated as 1. - hdds.heartbeat.max.command.status.report.size + hdds.heartbeat.command.status.report.max.size 64MB DATANODE Maximum serialized size of command status reports included in one heartbeat. The effective limit can be lower when other heartbeat fields consume the safe RPC payload budget. Reports beyond the limit stay queued - for a later heartbeat. + for a later heartbeat. A single command status that exceeds this limit is + sent as the only command status report when it still fits within the + remaining safe RPC payload, ensuring that it does not block all later + command statuses. diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java index fd5a4a3d7846..6165c497db95 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeQueueMetrics.java @@ -65,8 +65,6 @@ public final class DatanodeQueueMetrics implements MetricsSource { "CommandStatusReportQueue"; public static final String COMMAND_STATUS_REPORT_SENT_PREFIX = "CommandStatusReportSent"; - public static final String COMMAND_STATUS_REPORT_DEFERRED_PREFIX = - "CommandStatusReportDeferred"; public static final String COMMAND_STATUS_REPORT_REQUEUED_PREFIX = "CommandStatusReportRequeued"; public static final String COMMAND_STATUS_REPORT_OVERSIZED_PREFIX = @@ -84,11 +82,9 @@ public final class DatanodeQueueMetrics implements MetricsSource { private Map pipelineActionQueueMap; private Map commandStatusReportQueueBytesMap; private Map commandStatusReportSentBytesMap; - private Map commandStatusReportDeferredBytesMap; private Map commandStatusReportRequeuedBytesMap; private Map oversizedCommandStatusReportBytesMap; private Map commandStatusReportSentBytes; - private Map commandStatusReportDeferredBytes; private Map commandStatusReportRequeuedBytes; private Map oversizedCommandStatusReportBytes; @@ -132,11 +128,9 @@ private void initializeQueues() { pipelineActionQueueMap = new HashMap<>(); commandStatusReportQueueBytesMap = new HashMap<>(); commandStatusReportSentBytesMap = new HashMap<>(); - commandStatusReportDeferredBytesMap = new HashMap<>(); commandStatusReportRequeuedBytesMap = new HashMap<>(); oversizedCommandStatusReportBytesMap = new HashMap<>(); commandStatusReportSentBytes = new HashMap<>(); - commandStatusReportDeferredBytes = new HashMap<>(); commandStatusReportRequeuedBytes = new HashMap<>(); oversizedCommandStatusReportBytes = new HashMap<>(); } @@ -191,11 +185,6 @@ public void getMetrics(MetricsCollector collector, boolean b) { builder.addCounter(entry.getValue(), commandStatusReportSentBytes.get(entry.getKey()).get()); } - for (Map.Entry entry: - commandStatusReportDeferredBytesMap.entrySet()) { - builder.addGauge(entry.getValue(), - commandStatusReportDeferredBytes.get(entry.getKey()).get()); - } for (Map.Entry entry: commandStatusReportRequeuedBytesMap.entrySet()) { builder.addCounter(entry.getValue(), @@ -231,9 +220,6 @@ public void addEndpoint(HostAndPort endpoint) { commandStatusReportSentBytesMap.computeIfAbsent(endpoint, k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_SENT_PREFIX, endpointName, "Command status report bytes sent to ")); - commandStatusReportDeferredBytesMap.computeIfAbsent(endpoint, - k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_DEFERRED_PREFIX, - endpointName, "Command status report bytes deferred for ")); commandStatusReportRequeuedBytesMap.computeIfAbsent(endpoint, k -> getByteMetricsInfo(COMMAND_STATUS_REPORT_REQUEUED_PREFIX, endpointName, "Command status report bytes requeued for ")); @@ -242,8 +228,6 @@ public void addEndpoint(HostAndPort endpoint) { endpointName, "Oversized command status report bytes for ")); commandStatusReportSentBytes.computeIfAbsent(endpoint, k -> new AtomicLong()); - commandStatusReportDeferredBytes.computeIfAbsent(endpoint, - k -> new AtomicLong()); commandStatusReportRequeuedBytes.computeIfAbsent(endpoint, k -> new AtomicLong()); oversizedCommandStatusReportBytes.computeIfAbsent(endpoint, @@ -256,11 +240,9 @@ public void removeEndpoint(HostAndPort endpoint) { pipelineActionQueueMap.remove(endpoint); commandStatusReportQueueBytesMap.remove(endpoint); commandStatusReportSentBytesMap.remove(endpoint); - commandStatusReportDeferredBytesMap.remove(endpoint); commandStatusReportRequeuedBytesMap.remove(endpoint); oversizedCommandStatusReportBytesMap.remove(endpoint); commandStatusReportSentBytes.remove(endpoint); - commandStatusReportDeferredBytes.remove(endpoint); commandStatusReportRequeuedBytes.remove(endpoint); oversizedCommandStatusReportBytes.remove(endpoint); } @@ -271,12 +253,6 @@ public void incrCommandStatusReportBytesSent( k -> new AtomicLong()).addAndGet(bytes); } - public void setCommandStatusReportBytesDeferred( - HostAndPort endpoint, long bytes) { - commandStatusReportDeferredBytes.computeIfAbsent(endpoint, - k -> new AtomicLong()).set(bytes); - } - public void incrCommandStatusReportBytesRequeued( HostAndPort endpoint, long bytes) { commandStatusReportRequeuedBytes.computeIfAbsent(endpoint, @@ -289,6 +265,24 @@ public void setOversizedCommandStatusReportBytes( k -> new AtomicLong()).set(bytes); } + @VisibleForTesting + public long getCommandStatusReportBytesSent(HostAndPort endpoint) { + AtomicLong value = commandStatusReportSentBytes.get(endpoint); + return value == null ? 0 : value.get(); + } + + @VisibleForTesting + public long getCommandStatusReportBytesRequeued(HostAndPort endpoint) { + AtomicLong value = commandStatusReportRequeuedBytes.get(endpoint); + return value == null ? 0 : value.get(); + } + + @VisibleForTesting + public long getOversizedCommandStatusReportBytes(HostAndPort endpoint) { + AtomicLong value = oversizedCommandStatusReportBytes.get(endpoint); + return value == null ? 0 : value.get(); + } + @VisibleForTesting public int getIncrementalReportsQueueMapSize() { return incrementalReportsQueueMap.size(); diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java index 49c216dcdfbb..7d3ff75eb89b 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java @@ -379,20 +379,9 @@ public void putBackReports(List reportsToPutBack, */ public List getAllAvailableReports( HostAndPort endpoint) { - return getAllAvailableReportsUpToLimit(endpoint, Integer.MAX_VALUE); - } - - /** - * Returns up to {@code maxLimit} available reports for the endpoint, - * removing them from the queue. Any reports beyond the limit stay queued - * and are returned by a later call; callers can use - * {@link #hasPendingReports(HostAndPort)} to detect that case. - * - * @return List of reports - */ - public List getAllAvailableReports( - HostAndPort endpoint, int maxLimit) { - return getAllAvailableReportsUpToLimit(endpoint, maxLimit); + List reports = getFullReports(endpoint, Integer.MAX_VALUE); + reports.addAll(getIncrementalReports(endpoint, Integer.MAX_VALUE - reports.size())); + return reports; } /** @@ -411,11 +400,13 @@ public List getAvailableNonCommandStatusReports( /** * Selects command status reports within both the report count and serialized * byte limits. An aggregate report may be split between complete command - * statuses, but an individual command status is never split. + * statuses, but an individual command status is never split. To guarantee + * progress, one status may exceed {@code maxBytes} when it still fits within + * {@code maxSingleStatusBytes}. */ public CommandStatusReportSelection getCommandStatusReports( HostAndPort endpoint, int maxLimit, long maxBytes, - long maxConfiguredBytes) { + long maxSingleStatusBytes, long oversizedThreshold) { List selected = new LinkedList<>(); long selectedBytes = 0; long oversizedBytes = 0; @@ -448,33 +439,34 @@ public CommandStatusReportSelection getCommandStatusReports( selectedBytes += getCommandStatusReportWireSize(split.getSelected()); iterator.set(split.getRemainder()); - } else if (getFirstCommandStatusWireSize(report) - > maxConfiguredBytes) { - oversizedBytes = getFirstCommandStatusWireSize(report); + } else { + long firstStatusBytes = getFirstCommandStatusWireSize(report); + if (selected.isEmpty() && firstStatusBytes <= maxSingleStatusBytes) { + if (report.getCmdStatusCount() == 0) { + selected.add(report); + selectedBytes += firstStatusBytes; + iterator.remove(); + } else { + CommandStatusReportSplit singleStatus = + splitCommandStatusReport(report, firstStatusBytes); + selected.add(singleStatus.getSelected()); + selectedBytes += firstStatusBytes; + if (singleStatus.getRemainder().getCmdStatusCount() == 0) { + iterator.remove(); + } else { + iterator.set(singleStatus.getRemainder()); + } + } + } + if (firstStatusBytes > oversizedThreshold) { + oversizedBytes = firstStatusBytes; + } } break; } } - return new CommandStatusReportSelection(selected, selectedBytes, - getCommandStatusReportQueueBytes(endpoint), oversizedBytes); - } - } - - /** - * Returns true if any reports are still queued for the endpoint. - */ - public boolean hasPendingReports(HostAndPort endpoint) { - synchronized (incrementalReportsQueue) { - List reportsForEndpoint = incrementalReportsQueue.get(endpoint); - if (reportsForEndpoint != null && !reportsForEndpoint.isEmpty()) { - return true; - } + return new CommandStatusReportSelection(selected, selectedBytes, oversizedBytes); } - Map fullReports = - isFullReportReadyToBeSent.get(endpoint); - return fullReports != null && fullReports.entrySet().stream() - .anyMatch(report -> report.getValue().get() - && type2Reports.get(report.getKey()).get() != null); } /** @@ -507,17 +499,6 @@ public ContainerReportsProto getFullContainerReportDiscardPendingICR() } } - @VisibleForTesting - List getAllAvailableReportsUpToLimit( - HostAndPort endpoint, - int limit) { - List reports = getFullReports(endpoint, limit); - List incrementalReports = getIncrementalReports(endpoint, - limit - reports.size()); // get all (MAX_VALUE) - reports.addAll(incrementalReports); - return reports; - } - List getIncrementalReports( HostAndPort endpoint, int maxLimit) { List reportsToReturn = new LinkedList<>(); @@ -628,15 +609,12 @@ public Map getCommandStatusReportQueueBytes() { public static final class CommandStatusReportSelection { private final List reports; private final long selectedBytes; - private final long deferredBytes; private final long oversizedBytes; - private CommandStatusReportSelection( - List reports, long selectedBytes, - long deferredBytes, long oversizedBytes) { + private CommandStatusReportSelection(List reports, long selectedBytes, + long oversizedBytes) { this.reports = reports; this.selectedBytes = selectedBytes; - this.deferredBytes = deferredBytes; this.oversizedBytes = oversizedBytes; } @@ -648,10 +626,6 @@ public long getSelectedBytes() { return selectedBytes; } - public long getDeferredBytes() { - return deferredBytes; - } - public long getOversizedBytes() { return oversizedBytes; } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java index f32ce41fa72a..11dfbde2f1f9 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java @@ -21,8 +21,8 @@ import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_ACTION_MAX_LIMIT_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_MISSED_COUNT_THRESHOLD; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_ADDRESS_REFRESH_MISSED_COUNT_THRESHOLD_DEFAULT; -import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE; -import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE_DEFAULT; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT; @@ -119,10 +119,10 @@ public HeartbeatEndpointTask(EndpointStateMachine rpcEndpoint, this.maxReportsPerHB = Math.max(1, conf.getInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT)); this.maxCommandStatusReportBytes = (long) conf.getStorageSize( - HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE, - HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE_DEFAULT, BYTES); + HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE, + HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE_DEFAULT, BYTES); Preconditions.checkArgument(maxCommandStatusReportBytes > 0, - HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE + " must be positive"); + HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE + " must be positive"); int ipcMaximumDataLength = conf.getInt( CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH_DEFAULT); @@ -174,8 +174,7 @@ public EndpointStateMachine.EndPointStates call() throws Exception { addContainerActions(requestBuilder); addPipelineActions(requestBuilder); addQueuedCommandCounts(requestBuilder); - commandStatusSelection = addCommandStatusReports( - requestBuilder, maxReportsPerHB - reportCount); + commandStatusSelection = addCommandStatusReports(requestBuilder, maxReportsPerHB - reportCount); SCMHeartbeatRequestProto request = requestBuilder.build(); LOG.debug("Sending heartbeat message : {}", request); SCMHeartbeatResponseProto response = rpcEndpoint.getEndPoint() @@ -272,12 +271,12 @@ private CommandStatusReportSelection addCommandStatusReports( long currentHeartbeatSize = requestBuilder.build().getSerializedSize(); long remainingSafeBytes = Math.max( 0, safeHeartbeatSize - currentHeartbeatSize); - long commandStatusReportBudget = Math.min( - maxCommandStatusReportBytes, remainingSafeBytes); + long commandStatusReportBudget = Math.min(maxCommandStatusReportBytes, remainingSafeBytes); + long oversizedThreshold = Math.min(maxCommandStatusReportBytes, safeHeartbeatSize); CommandStatusReportSelection selection = context.getCommandStatusReports(rpcEndpoint.getAddress(), remainingReportCount, commandStatusReportBudget, - maxCommandStatusReportBytes); + remainingSafeBytes, oversizedThreshold); for (CommandStatusReportsProto report : selection.getReports()) { requestBuilder.addCommandStatusReports(report); } @@ -288,14 +287,12 @@ private CommandStatusReportSelection addCommandStatusReports( rpcEndpoint.getAddress(), currentHeartbeatSize, safeHeartbeatSize); } if (selection.getOversizedBytes() > 0) { - LOG.warn("A command status for {} is {} bytes, exceeding the configured " - + "per-heartbeat command status report limit of {} bytes", + LOG.warn("A command status for {} is {} bytes, exceeding the effective " + + "limit of {} bytes (configured limit: {}, safe heartbeat size: {})", rpcEndpoint.getAddress(), selection.getOversizedBytes(), - maxCommandStatusReportBytes); + oversizedThreshold, maxCommandStatusReportBytes, safeHeartbeatSize); } if (context.getQueueMetrics() != null) { - context.getQueueMetrics().setCommandStatusReportBytesDeferred( - rpcEndpoint.getAddress(), selection.getDeferredBytes()); context.getQueueMetrics().setOversizedCommandStatusReportBytes( rpcEndpoint.getAddress(), selection.getOversizedBytes()); } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java index 852cefad118a..6341c038fef3 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java @@ -232,27 +232,6 @@ public void testReportQueueWithAddReports() throws IOException { StateContext.CONTAINER_REPORTS_PROTO_NAME); } - @Test - public void testGetAllAvailableReportsRespectsLimit() throws IOException { - StateContext ctx = createSubject(); - HostAndPort scm1 = new HostAndPort("scm1", 9001); - ctx.addEndpoint(scm1); - - // Queue 10 ICRs; no full report is refreshed, so only ICRs are returned. - batchAddIncrementalReport(ctx, - StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME, 10); - - // A limited collection returns at most the limit and leaves the rest. - assertEquals(4, ctx.getAllAvailableReports(scm1, 4).size()); - assertTrue(ctx.hasPendingReports(scm1)); - assertEquals(4, ctx.getAllAvailableReports(scm1, 4).size()); - assertTrue(ctx.hasPendingReports(scm1)); - // Last chunk drains the queue. - assertEquals(2, ctx.getAllAvailableReports(scm1, 4).size()); - assertFalse(ctx.hasPendingReports(scm1)); - assertEquals(0, ctx.getAllAvailableReports(scm1, 4).size()); - } - @Test public void testCommandStatusReportsRespectByteLimit() throws IOException { StateContext context = createSubject(); @@ -265,17 +244,15 @@ public void testCommandStatusReportsRespectByteLimit() throws IOException { context.addIncrementalReport(second); long firstBytes = commandStatusReportWireSize(first); - CommandStatusReportSelection selection = context.getCommandStatusReports( - scm, 10, firstBytes, firstBytes); + CommandStatusReportSelection selection = + context.getCommandStatusReports(scm, 10, firstBytes, firstBytes, firstBytes); assertEquals(Collections.singletonList(first), selection.getReports()); assertEquals(firstBytes, selection.getSelectedBytes()); - assertEquals(commandStatusReportWireSize(second), - selection.getDeferredBytes()); + assertEquals(commandStatusReportWireSize(second), context.getCommandStatusReportQueueBytes(scm)); } @Test - public void testAggregateCommandStatusReportIsSplitAtStatusBoundary() - throws IOException { + public void testAggregateCommandStatusReportIsSplitAtStatusBoundary() throws IOException { StateContext context = createSubject(); HostAndPort scm = new HostAndPort("scm", 9001); context.addEndpoint(scm); @@ -302,23 +279,23 @@ public void testAggregateCommandStatusReportIsSplitAtStatusBoundary() CommandStatusReportSelection selection = context.getCommandStatusReports( scm, 10, commandStatusReportWireSize(firstBatch), + commandStatusReportWireSize(firstBatch), commandStatusReportWireSize(firstBatch)); assertEquals(Collections.singletonList(firstBatch), selection.getReports()); - assertEquals(commandStatusReportWireSize(secondBatch), - selection.getDeferredBytes()); + assertEquals(commandStatusReportWireSize(secondBatch), context.getCommandStatusReportQueueBytes(scm)); selection = context.getCommandStatusReports( scm, 10, commandStatusReportWireSize(firstBatch), + commandStatusReportWireSize(firstBatch), commandStatusReportWireSize(firstBatch)); assertEquals(Collections.singletonList(secondBatch), selection.getReports()); - assertEquals(0, selection.getDeferredBytes()); + assertEquals(0, context.getCommandStatusReportQueueBytes(scm)); } @Test - public void testNonCommandReportsAreSelectedIndependently() - throws IOException { + public void testNonCommandReportsAreSelectedIndependently() throws IOException { StateContext context = createSubject(); HostAndPort scm = new HostAndPort("scm", 9001); context.addEndpoint(scm); @@ -336,7 +313,7 @@ public void testNonCommandReportsAreSelectedIndependently() } @Test - public void testOversizedCommandStatusRemainsQueued() throws IOException { + public void testOversizedCommandStatusMakesProgressWhenSafe() throws IOException { StateContext context = createSubject(); HostAndPort scm = new HostAndPort("scm", 9001); context.addEndpoint(scm); @@ -344,38 +321,18 @@ public void testOversizedCommandStatusRemainsQueued() throws IOException { context.addIncrementalReport(report); long reportBytes = commandStatusReportWireSize(report); - CommandStatusReportSelection selection = context.getCommandStatusReports( - scm, 10, reportBytes - 1, reportBytes - 1); - assertTrue(selection.getReports().isEmpty()); - assertEquals(reportBytes, selection.getDeferredBytes()); + CommandStatusReportSelection selection = + context.getCommandStatusReports(scm, 10, reportBytes - 1, reportBytes, reportBytes - 1); + assertEquals(Collections.singletonList(report), selection.getReports()); + assertEquals(0, context.getCommandStatusReportQueueBytes(scm)); assertEquals(reportBytes, selection.getOversizedBytes()); + context.addIncrementalReport(report); selection = context.getCommandStatusReports( - scm, 10, reportBytes, reportBytes); - assertEquals(Collections.singletonList(report), selection.getReports()); - assertEquals(0, selection.getDeferredBytes()); - } - - @Test - public void testHasPendingReportsWithFullReports() throws IOException { - StateContext ctx = createSubject(); - HostAndPort scm1 = new HostAndPort("scm1", 9001); - ctx.addEndpoint(scm1); - - // addEndpoint marks the full-report ready flags true, but no full report - // value has been generated yet, so nothing is actually pending. - assertFalse(ctx.hasPendingReports(scm1)); - - // Generate all three full reports; now values exist and flags are ready. - batchRefreshfullReports(ctx, StateContext.CONTAINER_REPORTS_PROTO_NAME, 1); - batchRefreshfullReports(ctx, StateContext.NODE_REPORT_PROTO_NAME, 1); - batchRefreshfullReports(ctx, StateContext.PIPELINE_REPORTS_PROTO_NAME, 1); - assertTrue(ctx.hasPendingReports(scm1)); - - // With a cap of one, a heartbeat takes a single full report but the others - // stay ready, so a follow-up is still needed. - assertEquals(1, ctx.getAllAvailableReports(scm1, 1).size()); - assertTrue(ctx.hasPendingReports(scm1)); + scm, 10, reportBytes - 1, reportBytes - 1, reportBytes - 1); + assertTrue(selection.getReports().isEmpty()); + assertEquals(reportBytes, context.getCommandStatusReportQueueBytes(scm)); + assertEquals(reportBytes, selection.getOversizedBytes()); } void batchRefreshfullReports(StateContext ctx, String reportName, int count) { @@ -860,17 +817,17 @@ public void testGetReports() { // (100 - 3 non-incremental reports) expectedReportCount.put( StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME, 97); - checkReportCount(ctx.getAllAvailableReportsUpToLimit(scm1, 100), + checkReportCount(ctx.getAvailableNonCommandStatusReports(scm1, 100), expectedReportCount); - checkReportCount(ctx.getAllAvailableReportsUpToLimit(scm2, 100), + checkReportCount(ctx.getAvailableNonCommandStatusReports(scm2, 100), expectedReportCount); expectedReportCount.clear(); expectedReportCount.put( StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME, totalIncrementalCount - 97); - checkReportCount(ctx.getAllAvailableReportsUpToLimit(scm1, 100), + checkReportCount(ctx.getAvailableNonCommandStatusReports(scm1, 100), expectedReportCount); - checkReportCount(ctx.getAllAvailableReportsUpToLimit(scm2, 100), + checkReportCount(ctx.getAvailableNonCommandStatusReports(scm2, 100), expectedReportCount); } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java index 2c6417fcc0ed..30e81e2d9860 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java @@ -18,7 +18,7 @@ package org.apache.hadoop.ozone.container.common.states.endpoint; import static java.util.Collections.emptyList; -import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT; import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type.reconcileContainerCommand; import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type.reconstructECContainersCommand; @@ -66,6 +66,7 @@ import org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature; import org.apache.hadoop.hdfs.util.EnumCounters; import org.apache.hadoop.ozone.container.common.ContainerTestUtils; +import org.apache.hadoop.ozone.container.common.statemachine.DatanodeQueueMetrics; import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine; import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine.DatanodeStates; import org.apache.hadoop.ozone.container.common.statemachine.EndpointStateMachine; @@ -414,8 +415,7 @@ public void testheartbeatWithAllReports() throws Exception { } @Test - public void testHeartbeatCapsReports() - throws Exception { + public void testHeartbeatCapsReports() throws Exception { OzoneConfiguration conf = new OzoneConfiguration(); conf.setInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, 2); DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); @@ -452,28 +452,28 @@ public void testHeartbeatCapsReports() SCMHeartbeatRequestProto heartbeat = argument.getValue(); assertEquals(2, heartbeat.getIncrementalContainerReportCount()); // The remainder stays queued for the next regular heartbeat. - assertTrue(context.hasPendingReports(TEST_SCM_ENDPOINT)); + assertEquals(3, context.getIncrementalReportQueueSize().get(TEST_SCM_ENDPOINT)); verify(datanodeStateMachine, never()).setNextHB(anyLong()); // Subsequent regular heartbeats drain the rest. endpointTask.call(); endpointTask.call(); assertEquals(1, argument.getValue().getIncrementalContainerReportCount()); - assertFalse(context.hasPendingReports(TEST_SCM_ENDPOINT)); + assertEquals(0, context.getIncrementalReportQueueSize().get(TEST_SCM_ENDPOINT)); verify(datanodeStateMachine, never()).setNextHB(anyLong()); } @Test - public void testHeartbeatCapsCommandStatusReportsByBytes() - throws Exception { + public void testHeartbeatCapsCommandStatusReportsByBytes() throws Exception { OzoneConfiguration conf = new OzoneConfiguration(); conf.setInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, 10); CommandStatusReportsProto first = newCommandStatusReport(1, 128); CommandStatusReportsProto second = newCommandStatusReport(2, 128); long reportBytes = commandStatusReportWireSize(first); - conf.set(HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE, - reportBytes + "B"); + conf.set(HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE, reportBytes + "B"); DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + DatanodeQueueMetrics queueMetrics = new DatanodeQueueMetrics(datanodeStateMachine); + when(datanodeStateMachine.getQueueMetrics()).thenReturn(queueMetrics); StateContext context = new StateContext(conf, DatanodeStates.RUNNING, datanodeStateMachine, ""); when(datanodeStateMachine.getQueuedCommandCount()) @@ -502,13 +502,84 @@ public void testHeartbeatCapsCommandStatusReportsByBytes() assertEquals(1, argument.getValue().getCommandStatusReportsCount()); assertEquals(first, argument.getValue().getCommandStatusReports(0)); - assertTrue(context.hasPendingReports(TEST_SCM_ENDPOINT)); + assertEquals(commandStatusReportWireSize(second), + context.getCommandStatusReportQueueBytes(TEST_SCM_ENDPOINT)); + assertEquals(reportBytes, queueMetrics.getCommandStatusReportBytesSent(TEST_SCM_ENDPOINT)); + assertEquals(0, queueMetrics.getOversizedCommandStatusReportBytes(TEST_SCM_ENDPOINT)); verify(datanodeStateMachine, never()).setNextHB(anyLong()); } @Test - public void testExistingHeartbeatContentReducesCommandStatusBudget() - throws Exception { + public void testNonCommandReportsConsumeReportCountLimit() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + conf.setInt(HDDS_HEARTBEAT_REPORT_MAX_LIMIT, 2); + DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, datanodeStateMachine, ""); + when(datanodeStateMachine.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + + StorageContainerDatanodeProtocolClientSideTranslatorPB scm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + ArgumentCaptor argument = ArgumentCaptor.forClass(SCMHeartbeatRequestProto.class); + when(scm.sendHeartbeat(argument.capture())) + .thenAnswer(invocation -> + SCMHeartbeatResponseProto.newBuilder() + .setDatanodeUUID(((SCMHeartbeatRequestProto) invocation.getArgument(0)) + .getDatanodeDetails().getUuid()) + .build()); + + HeartbeatEndpointTask endpointTask = getHeartbeatEndpointTask(conf, context, scm); + context.addEndpoint(TEST_SCM_ENDPOINT); + context.addIncrementalReport(IncrementalContainerReportProto.getDefaultInstance()); + context.addIncrementalReport(IncrementalContainerReportProto.getDefaultInstance()); + CommandStatusReportsProto commandStatus = newCommandStatusReport(1, 128); + context.addIncrementalReport(commandStatus); + + endpointTask.call(); + + assertEquals(2, argument.getValue().getIncrementalContainerReportCount()); + assertEquals(0, argument.getValue().getCommandStatusReportsCount()); + assertEquals(commandStatusReportWireSize(commandStatus), + context.getCommandStatusReportQueueBytes(TEST_SCM_ENDPOINT)); + } + + @Test + public void testOversizedCommandStatusSentWhenSafe() throws Exception { + CommandStatusReportsProto report = newCommandStatusReport(1, 128); + long reportBytes = commandStatusReportWireSize(report); + OzoneConfiguration conf = new OzoneConfiguration(); + conf.set(HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE, (reportBytes - 1) + "B"); + DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + DatanodeQueueMetrics queueMetrics = new DatanodeQueueMetrics(datanodeStateMachine); + when(datanodeStateMachine.getQueueMetrics()).thenReturn(queueMetrics); + when(datanodeStateMachine.getQueuedCommandCount()) + .thenReturn(new EnumCounters<>(SCMCommandProto.Type.class)); + StateContext context = new StateContext(conf, DatanodeStates.RUNNING, datanodeStateMachine, ""); + + StorageContainerDatanodeProtocolClientSideTranslatorPB scm = + mock(StorageContainerDatanodeProtocolClientSideTranslatorPB.class); + ArgumentCaptor argument = ArgumentCaptor.forClass(SCMHeartbeatRequestProto.class); + when(scm.sendHeartbeat(argument.capture())) + .thenAnswer(invocation -> + SCMHeartbeatResponseProto.newBuilder() + .setDatanodeUUID(((SCMHeartbeatRequestProto) invocation.getArgument(0)) + .getDatanodeDetails().getUuid()) + .build()); + + HeartbeatEndpointTask endpointTask = getHeartbeatEndpointTask(conf, context, scm); + context.addEndpoint(TEST_SCM_ENDPOINT); + context.addIncrementalReport(report); + + endpointTask.call(); + + assertEquals(report, argument.getValue().getCommandStatusReports(0)); + assertEquals(0, context.getCommandStatusReportQueueBytes(TEST_SCM_ENDPOINT)); + assertEquals(reportBytes, queueMetrics.getCommandStatusReportBytesSent(TEST_SCM_ENDPOINT)); + assertEquals(reportBytes, queueMetrics.getOversizedCommandStatusReportBytes(TEST_SCM_ENDPOINT)); + } + + @Test + public void testExistingHeartbeatContentReducesCommandStatusBudget() throws Exception { IncrementalContainerReportProto icr = IncrementalContainerReportProto.newBuilder() .addReport(ContainerReplicaProto.newBuilder() @@ -555,8 +626,7 @@ public void testExistingHeartbeatContentReducesCommandStatusBudget() OzoneConfiguration conf = new OzoneConfiguration(); conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, ipcMaximumDataLength); - conf.set(HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE, - (reportBytes * 2) + "B"); + conf.set(HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE, (reportBytes * 2) + "B"); StateContext context = new StateContext(conf, DatanodeStates.RUNNING, datanodeStateMachine, ""); StorageContainerDatanodeProtocolClientSideTranslatorPB scm = @@ -581,19 +651,20 @@ public void testExistingHeartbeatContentReducesCommandStatusBudget() assertEquals(1, argument.getValue().getCommandStatusReportsCount()); assertTrue(argument.getValue().getSerializedSize() <= safeHeartbeatSize); - assertTrue(context.hasPendingReports(TEST_SCM_ENDPOINT)); + assertEquals(commandStatusReportWireSize(second), + context.getCommandStatusReportQueueBytes(TEST_SCM_ENDPOINT)); } @Test - public void testFailedHeartbeatRestoresCommandStatusReports() - throws Exception { + public void testFailedHeartbeatRestoresCommandStatusReports() throws Exception { OzoneConfiguration conf = new OzoneConfiguration(); CommandStatusReportsProto first = newCommandStatusReport(1, 128); CommandStatusReportsProto second = newCommandStatusReport(2, 128); long reportBytes = commandStatusReportWireSize(first); - conf.set(HDDS_HEARTBEAT_MAX_COMMAND_STATUS_REPORT_SIZE, - reportBytes + "B"); + conf.set(HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE, reportBytes + "B"); DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); + DatanodeQueueMetrics queueMetrics = new DatanodeQueueMetrics(datanodeStateMachine); + when(datanodeStateMachine.getQueueMetrics()).thenReturn(queueMetrics); StateContext context = new StateContext(conf, DatanodeStates.RUNNING, datanodeStateMachine, ""); when(datanodeStateMachine.getQueuedCommandCount()) @@ -612,8 +683,9 @@ public void testFailedHeartbeatRestoresCommandStatusReports() endpointTask.call(); CommandStatusReportSelection selection = context.getCommandStatusReports( - TEST_SCM_ENDPOINT, 10, reportBytes * 2, reportBytes * 2); + TEST_SCM_ENDPOINT, 10, reportBytes * 2, reportBytes * 2, reportBytes * 2); assertEquals(Arrays.asList(first, second), selection.getReports()); + assertEquals(reportBytes, queueMetrics.getCommandStatusReportBytesRequeued(TEST_SCM_ENDPOINT)); } @ParameterizedTest @@ -653,7 +725,7 @@ public void testHeartbeatClampsNonPositiveReportLimit(int configuredLimit) // progress (one ICR) rather than sending nothing (0) or throwing (-1). endpointTask.call(); assertEquals(1, argument.getValue().getIncrementalContainerReportCount()); - assertTrue(context.hasPendingReports(TEST_SCM_ENDPOINT)); + assertEquals(2, context.getIncrementalReportQueueSize().get(TEST_SCM_ENDPOINT)); } @Test From 877f6c3c97d93df2fbe34bc6eea73d867297f23a Mon Sep 17 00:00:00 2001 From: sadanand48 Date: Tue, 10 Nov 2026 11:10:52 +0530 Subject: [PATCH 4/8] address comments --- .../org/apache/hadoop/hdds/scm/ScmConfig.java | 6 +-- .../common/statemachine/TestStateContext.java | 42 +++++++++++++++++++ 2 files changed, 44 insertions(+), 4 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java index 0ddf71f9eeee..7cdb81116c0d 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfig.java @@ -114,10 +114,8 @@ public class ScmConfig extends ReconfigurableConfig { + "configuration is 5000 then it would only send 5000 blocks " + "for deletion to the datanodes. Increasing this value can " + "increase delete command acknowledgement size; each " - + "acknowledgement must fit within the datanode heartbeat " - + "command status report limit " - + "(hdds.heartbeat.command.status.report.max.size) and the IPC " - + "size limit (ipc.maximum.data.length)." + + "individual acknowledgement must fit within the IPC size " + + "limit (ipc.maximum.data.length)." ) private int blockDeletionLimit; diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java index 6341c038fef3..76569663d518 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java @@ -335,6 +335,48 @@ public void testOversizedCommandStatusMakesProgressWhenSafe() throws IOException assertEquals(reportBytes, selection.getOversizedBytes()); } + @Test + public void testOversizedCommandStatusPreservesAggregateOrder() throws IOException { + StateContext context = createSubject(); + HostAndPort scm = new HostAndPort("scm", 9001); + context.addEndpoint(scm); + CommandStatus firstSmall = newCommandStatus(1, 16); + CommandStatus oversized = newCommandStatus(2, 256); + CommandStatus lastSmall = newCommandStatus(3, 16); + CommandStatusReportsProto aggregate = CommandStatusReportsProto.newBuilder() + .addCmdStatus(firstSmall) + .addCmdStatus(oversized) + .addCmdStatus(lastSmall) + .build(); + CommandStatusReportsProto firstSmallReport = + CommandStatusReportsProto.newBuilder().addCmdStatus(firstSmall).build(); + CommandStatusReportsProto oversizedReport = + CommandStatusReportsProto.newBuilder().addCmdStatus(oversized).build(); + CommandStatusReportsProto lastSmallReport = + CommandStatusReportsProto.newBuilder().addCmdStatus(lastSmall).build(); + CommandStatusReportsProto firstRemainder = CommandStatusReportsProto.newBuilder() + .addCmdStatus(oversized) + .addCmdStatus(lastSmall) + .build(); + long smallBytes = commandStatusReportWireSize(firstSmallReport); + long oversizedBytes = commandStatusReportWireSize(oversizedReport); + context.addIncrementalReport(aggregate); + + CommandStatusReportSelection selection = + context.getCommandStatusReports(scm, 10, smallBytes, oversizedBytes, smallBytes); + assertEquals(Collections.singletonList(firstSmallReport), selection.getReports()); + assertEquals(commandStatusReportWireSize(firstRemainder), context.getCommandStatusReportQueueBytes(scm)); + + selection = context.getCommandStatusReports(scm, 10, smallBytes, oversizedBytes, smallBytes); + assertEquals(Collections.singletonList(oversizedReport), selection.getReports()); + assertEquals(oversizedBytes, selection.getOversizedBytes()); + assertEquals(commandStatusReportWireSize(lastSmallReport), context.getCommandStatusReportQueueBytes(scm)); + + selection = context.getCommandStatusReports(scm, 10, smallBytes, oversizedBytes, smallBytes); + assertEquals(Collections.singletonList(lastSmallReport), selection.getReports()); + assertEquals(0, context.getCommandStatusReportQueueBytes(scm)); + } + void batchRefreshfullReports(StateContext ctx, String reportName, int count) { for (int i = 0; i < count; i++) { ctx.refreshFullReport(newMockReport(reportName)); From f61c2c56df10b9553f4c79d3e912a21f53ba3786 Mon Sep 17 00:00:00 2001 From: sadanand48 Date: Wed, 11 Nov 2026 11:49:57 +0530 Subject: [PATCH 5/8] address comments --- .../endpoint/HeartbeatEndpointTask.java | 53 +++++++++++++------ 1 file changed, 37 insertions(+), 16 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java index 11dfbde2f1f9..e35247e35ecf 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java @@ -33,6 +33,7 @@ import static org.apache.hadoop.ozone.container.upgrade.UpgradeUtils.toVersionProto; import com.google.common.base.Preconditions; +import com.google.protobuf.CodedOutputStream; import com.google.protobuf.Descriptors; import com.google.protobuf.Message; import java.io.IOException; @@ -170,11 +171,15 @@ public EndpointStateMachine.EndPointStates call() throws Exception { requestBuilder = SCMHeartbeatRequestProto.newBuilder() .setDatanodeDetails(datanodeDetailsProto) .setDatanodeVersion(versionInfo); - int reportCount = addNonCommandStatusReports(requestBuilder); - addContainerActions(requestBuilder); - addPipelineActions(requestBuilder); - addQueuedCommandCounts(requestBuilder); - commandStatusSelection = addCommandStatusReports(requestBuilder, maxReportsPerHB - reportCount); + SerializedSizeAccumulator heartbeatSize = new SerializedSizeAccumulator(); + heartbeatSize.add(SCMHeartbeatRequestProto.DATANODEDETAILS_FIELD_NUMBER, datanodeDetailsProto); + heartbeatSize.add(SCMHeartbeatRequestProto.DATANODEVERSION_FIELD_NUMBER, versionInfo); + int reportCount = addNonCommandStatusReports(requestBuilder, heartbeatSize); + addContainerActions(requestBuilder, heartbeatSize); + addPipelineActions(requestBuilder, heartbeatSize); + addQueuedCommandCounts(requestBuilder, heartbeatSize); + commandStatusSelection = addCommandStatusReports( + requestBuilder, maxReportsPerHB - reportCount, heartbeatSize.get()); SCMHeartbeatRequestProto request = requestBuilder.build(); LOG.debug("Sending heartbeat message : {}", request); SCMHeartbeatResponseProto response = rpcEndpoint.getEndPoint() @@ -241,8 +246,8 @@ private void putBackIncrementalReports( context.putBackReports(reports, rpcEndpoint.getAddress()); } - private int addNonCommandStatusReports( - SCMHeartbeatRequestProto.Builder requestBuilder) { + private int addNonCommandStatusReports(SCMHeartbeatRequestProto.Builder requestBuilder, + SerializedSizeAccumulator heartbeatSize) { int reportCount = 0; for (Message report : context.getAvailableNonCommandStatusReports( @@ -258,6 +263,7 @@ private int addNonCommandStatusReports( } else { requestBuilder.setField(descriptor, report); } + heartbeatSize.add(descriptor.getNumber(), report); break; } } @@ -267,8 +273,7 @@ private int addNonCommandStatusReports( private CommandStatusReportSelection addCommandStatusReports( SCMHeartbeatRequestProto.Builder requestBuilder, - int remainingReportCount) { - long currentHeartbeatSize = requestBuilder.build().getSerializedSize(); + int remainingReportCount, long currentHeartbeatSize) { long remainingSafeBytes = Math.max( 0, safeHeartbeatSize - currentHeartbeatSize); long commandStatusReportBudget = Math.min(maxCommandStatusReportBytes, remainingSafeBytes); @@ -304,8 +309,8 @@ private CommandStatusReportSelection addCommandStatusReports( * * @param requestBuilder builder to which the report has to be added. */ - private void addContainerActions( - SCMHeartbeatRequestProto.Builder requestBuilder) { + private void addContainerActions(SCMHeartbeatRequestProto.Builder requestBuilder, + SerializedSizeAccumulator heartbeatSize) { List actions = context.getPendingContainerAction( rpcEndpoint.getAddress(), maxContainerActionsPerHB); if (!actions.isEmpty()) { @@ -313,6 +318,7 @@ private void addContainerActions( .addAllContainerActions(actions) .build(); requestBuilder.setContainerActions(cap); + heartbeatSize.add(SCMHeartbeatRequestProto.CONTAINERACTIONS_FIELD_NUMBER, cap); } } @@ -321,8 +327,8 @@ private void addContainerActions( * * @param requestBuilder builder to which the report has to be added. */ - private void addPipelineActions( - SCMHeartbeatRequestProto.Builder requestBuilder) { + private void addPipelineActions(SCMHeartbeatRequestProto.Builder requestBuilder, + SerializedSizeAccumulator heartbeatSize) { List actions = context.getPendingPipelineAction( rpcEndpoint.getAddress(), maxPipelineActionsPerHB); if (!actions.isEmpty()) { @@ -330,6 +336,7 @@ private void addPipelineActions( .addAllPipelineActions(actions) .build(); requestBuilder.setPipelineActions(pap); + heartbeatSize.add(SCMHeartbeatRequestProto.PIPELINEACTIONS_FIELD_NUMBER, pap); } } @@ -337,8 +344,8 @@ private void addPipelineActions( * Adds the count of all queued commands to the heartbeat. * @param requestBuilder Builder to which the details will be added. */ - private void addQueuedCommandCounts( - SCMHeartbeatRequestProto.Builder requestBuilder) { + private void addQueuedCommandCounts(SCMHeartbeatRequestProto.Builder requestBuilder, + SerializedSizeAccumulator heartbeatSize) { EnumCounters commandCount = context.getParent().getQueuedCommandCount(); CommandQueueReportProto.Builder reportProto = @@ -350,7 +357,21 @@ private void addQueuedCommandCounts( .addCount((int) count); } } - requestBuilder.setCommandQueueReport(reportProto.build()); + CommandQueueReportProto commandQueueReport = reportProto.build(); + requestBuilder.setCommandQueueReport(commandQueueReport); + heartbeatSize.add(SCMHeartbeatRequestProto.COMMANDQUEUEREPORT_FIELD_NUMBER, commandQueueReport); + } + + private static final class SerializedSizeAccumulator { + private long size; + + private void add(int fieldNumber, Message message) { + size += CodedOutputStream.computeMessageSize(fieldNumber, message); + } + + private long get() { + return size; + } } /** From bcfc1b29cab58dd997dd49232d7dd881e6bfbb1c Mon Sep 17 00:00:00 2001 From: "Doroszlai, Attila" <6454655+adoroszlai@users.noreply.github.com> Date: Fri, 13 Nov 2026 09:45:50 +0100 Subject: [PATCH 6/8] trivial formatting change --- .../common/states/endpoint/TestHeartbeatEndpointTask.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java index 30e81e2d9860..4b709265ab82 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java @@ -585,8 +585,7 @@ public void testExistingHeartbeatContentReducesCommandStatusBudget() throws Exce .addReport(ContainerReplicaProto.newBuilder() .setContainerID(1) .setState(ContainerReplicaProto.State.CLOSED) - .setOriginNodeId(String.join("", Collections.nCopies( - 1024, "x"))) + .setOriginNodeId(String.join("", Collections.nCopies(1024, "x"))) .build()) .build(); DatanodeStateMachine datanodeStateMachine = mockDatanodeStateMachine(); From 853c5f81c22eb9382df213c3cc08733401612806 Mon Sep 17 00:00:00 2001 From: sadanand48 Date: Fri, 13 Nov 2026 14:35:02 +0530 Subject: [PATCH 7/8] checkstyle --- .../main/java/org/apache/hadoop/hdds/HddsConfigKeys.java | 4 ++++ .../container/common/statemachine/StateContext.java | 3 +++ .../common/states/endpoint/HeartbeatEndpointTask.java | 9 +++++---- .../states/endpoint/TestHeartbeatEndpointTask.java | 5 ++--- 4 files changed, 14 insertions(+), 7 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java index 5e44f1ee19c0..3fb580c4f6b6 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java @@ -76,6 +76,10 @@ public final class HddsConfigKeys { public static final String HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE = "hdds.heartbeat.command.status.report.max.size"; public static final String HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE_DEFAULT = "64MB"; + public static final String HDDS_IPC_MAXIMUM_DATA_LENGTH = + "ipc.maximum.data.length"; + public static final int HDDS_IPC_MAXIMUM_DATA_LENGTH_DEFAULT = + 128 * 1024 * 1024; // Configuration to allow volume choosing policy. public static final String HDDS_DATANODE_VOLUME_CHOOSING_POLICY = "hdds.datanode.volume.choosing.policy"; diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java index 7d3ff75eb89b..d40e1e3a7c05 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java @@ -606,6 +606,9 @@ public Map getCommandStatusReportQueueBytes() { } } + /** + * Command status reports selected for one heartbeat and their byte metrics. + */ public static final class CommandStatusReportSelection { private final List reports; private final long selectedBytes; diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java index e35247e35ecf..2b6a6112b61a 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java @@ -25,6 +25,8 @@ import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_IPC_MAXIMUM_DATA_LENGTH; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_IPC_MAXIMUM_DATA_LENGTH_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT_DEFAULT; import static org.apache.hadoop.hdds.conf.StorageUnit.BYTES; @@ -41,7 +43,6 @@ import java.util.LinkedList; import java.util.List; import java.util.concurrent.Callable; -import org.apache.hadoop.fs.CommonConfigurationKeys; import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.protocol.DatanodeDetails; @@ -125,10 +126,10 @@ public HeartbeatEndpointTask(EndpointStateMachine rpcEndpoint, Preconditions.checkArgument(maxCommandStatusReportBytes > 0, HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE + " must be positive"); int ipcMaximumDataLength = conf.getInt( - CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, - CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH_DEFAULT); + HDDS_IPC_MAXIMUM_DATA_LENGTH, + HDDS_IPC_MAXIMUM_DATA_LENGTH_DEFAULT); Preconditions.checkArgument(ipcMaximumDataLength > 0, - CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH + " must be positive"); + HDDS_IPC_MAXIMUM_DATA_LENGTH + " must be positive"); this.safeHeartbeatSize = (long) Math.floor(ipcMaximumDataLength * IPC_SIZE_SAFETY_FACTOR); this.datanodeDetails = context.getParent().getDatanodeDetails(); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java index 4b709265ab82..4312bf94f8e2 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java @@ -20,6 +20,7 @@ import static java.util.Collections.emptyList; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT; +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_IPC_MAXIMUM_DATA_LENGTH; import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type.reconcileContainerCommand; import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type.reconstructECContainersCommand; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -44,7 +45,6 @@ import java.util.OptionalLong; import java.util.Set; import java.util.UUID; -import org.apache.hadoop.fs.CommonConfigurationKeys; import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.conf.ConfigurationSource; @@ -623,8 +623,7 @@ public void testExistingHeartbeatContentReducesCommandStatusBudget() throws Exce long safeHeartbeatSize = (long) Math.floor(ipcMaximumDataLength * 0.9D); OzoneConfiguration conf = new OzoneConfiguration(); - conf.setInt(CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH, - ipcMaximumDataLength); + conf.setInt(HDDS_IPC_MAXIMUM_DATA_LENGTH, ipcMaximumDataLength); conf.set(HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE, (reportBytes * 2) + "B"); StateContext context = new StateContext(conf, DatanodeStates.RUNNING, datanodeStateMachine, ""); From d35de55c4c925fea5e5c121fa2e994de1fefeb17 Mon Sep 17 00:00:00 2001 From: sadanand48 Date: Fri, 13 Nov 2026 15:31:01 +0530 Subject: [PATCH 8/8] move config --- .../main/java/org/apache/hadoop/hdds/HddsConfigKeys.java | 4 ---- .../common/states/endpoint/HeartbeatEndpointTask.java | 6 ++++-- .../common/states/endpoint/TestHeartbeatEndpointTask.java | 2 +- 3 files changed, 5 insertions(+), 7 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java index 3fb580c4f6b6..5e44f1ee19c0 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsConfigKeys.java @@ -76,10 +76,6 @@ public final class HddsConfigKeys { public static final String HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE = "hdds.heartbeat.command.status.report.max.size"; public static final String HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE_DEFAULT = "64MB"; - public static final String HDDS_IPC_MAXIMUM_DATA_LENGTH = - "ipc.maximum.data.length"; - public static final int HDDS_IPC_MAXIMUM_DATA_LENGTH_DEFAULT = - 128 * 1024 * 1024; // Configuration to allow volume choosing policy. public static final String HDDS_DATANODE_VOLUME_CHOOSING_POLICY = "hdds.datanode.volume.choosing.policy"; diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java index 2b6a6112b61a..a093066e7697 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java @@ -25,8 +25,6 @@ import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT_DEFAULT; -import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_IPC_MAXIMUM_DATA_LENGTH; -import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_IPC_MAXIMUM_DATA_LENGTH_DEFAULT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_PIPELINE_ACTION_MAX_LIMIT_DEFAULT; import static org.apache.hadoop.hdds.conf.StorageUnit.BYTES; @@ -88,6 +86,10 @@ public class HeartbeatEndpointTask implements Callable { private static final Logger LOG = LoggerFactory.getLogger(HeartbeatEndpointTask.class); + static final String HDDS_IPC_MAXIMUM_DATA_LENGTH = + "ipc.maximum.data.length"; + static final int HDDS_IPC_MAXIMUM_DATA_LENGTH_DEFAULT = + 128 * 1024 * 1024; private static final double IPC_SIZE_SAFETY_FACTOR = 0.9; private final EndpointStateMachine rpcEndpoint; private final DatanodeDetails datanodeDetails; diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java index 4312bf94f8e2..815a3bf97910 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/states/endpoint/TestHeartbeatEndpointTask.java @@ -20,9 +20,9 @@ import static java.util.Collections.emptyList; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_COMMAND_STATUS_REPORT_MAX_SIZE; import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_REPORT_MAX_LIMIT; -import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_IPC_MAXIMUM_DATA_LENGTH; import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type.reconcileContainerCommand; import static org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type.reconstructECContainersCommand; +import static org.apache.hadoop.ozone.container.common.states.endpoint.HeartbeatEndpointTask.HDDS_IPC_MAXIMUM_DATA_LENGTH; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals;