Merge branch 'apache:master' into HDDS-15172

Author: slfan1989

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/tracing/TracingUtil.java

@@ -26,6 +26,7 @@
 import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.Scope;
+import io.opentelemetry.context.propagation.TextMapGetter;
 import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
 import io.opentelemetry.sdk.resources.Resource;
@@ -36,6 +37,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.function.Function;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.ratis.util.function.CheckedRunnable;
 import org.apache.ratis.util.function.CheckedSupplier;
@@ -387,4 +389,46 @@ private static Span buildSpan(String spanName) {
       return tracer.spanBuilder(spanName).setNoParent().startSpan();
     }
   }
+
+  /**
+   * A TextMapGetter implementation to extract tracing info from getHeader.
+   */
+  public static class HttpHeaderGetter implements TextMapGetter<Function<String, String>> {
+
+    @Override
+    public Iterable<String> keys(Function<String, String> carrier) {
+      // Not used during the extract call, so returning an empty list.
+      return Collections.emptyList();
+    }
+
+    @Override
+    public String get(Function<String, String> carrier, String key) {
+      return carrier == null ? null : carrier.apply(key);
+    }
+  }
+
+  public static TraceCloseable createActivatedSpanFromW3cHttpHeaders(
+      String spanName, Function<String, String> getHeader, ConfigurationSource conf) {
+    if (conf == null || !isTracingEnabled(conf)) {
+      return () -> { };
+    }
+
+    Context remote = W3CTraceContextPropagator.getInstance()
+        .extract(Context.current(), getHeader, new HttpHeaderGetter());
+
+    if (!Span.fromContext(remote).getSpanContext().isValid()) {
+      return createActivatedSpan(spanName);
+    }
+
+    Span span = tracer.spanBuilder(spanName)
+        .setParent(remote)
+        .startSpan();
+
+    Scope scope = span.makeCurrent();
+
+    return () -> {
+      scope.close();
+      span.end();
+    };
+  }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerSet.java

@@ -63,6 +63,12 @@ public class ContainerSet implements Iterable<Container<?>> {
 
   private static final Logger LOG = LoggerFactory.getLogger(ContainerSet.class);
 
+  /**
+   * Max attempts to acquire {@link Container#writeLock()} while verifying this set's id → container mapping
+   * is same (e.g. another thread may {@link #updateContainer} / DiskBalancer swap the instance).
+   */
+  private static final int MAX_CONTAINER_MAP_SWAP_RETRIES = 5;
+
   private final ConcurrentSkipListMap<Long, Container<?>> containerMap = new
       ConcurrentSkipListMap<>();
   private final ConcurrentSkipListSet<Long> missingContainerSet =
@@ -279,6 +285,54 @@ public Container<?> getContainer(long containerId) {
     return containerMap.get(containerId);
   }
 
+  /**
+   * Returns the max retry for a container map swap while acquiring container lock.
+   * @return max retry count
+   */
+  public static int maxContainerMapSwapRetries() {
+    return MAX_CONTAINER_MAP_SWAP_RETRIES;
+  }
+
+  /**
+   * Locks the container mapped to {@code containerId} for write, and verifies that the instance locked is still
+   * the one stored in this set. If the mapping is swapped, unlocks and retries up to
+   * {@link #maxContainerMapSwapRetries()} times, then returns {@code null}.
+   *
+   * @return the locked container, or {@code null} if the mapping could not be stabilized after all retries
+   * @throws StorageContainerException with {@code CONTAINER_NOT_FOUND}
+   */
+  @Nullable
+  public Container<?> getContainerWithWriteLock(long containerId) throws StorageContainerException {
+    for (int retry = 0; retry < MAX_CONTAINER_MAP_SWAP_RETRIES; retry++) {
+      Container<?> candidate = getContainer(containerId);
+      if (candidate == null) {
+        throw new StorageContainerException(
+            "Container " + containerId + " not found in ContainerSet.",
+            ContainerProtos.Result.CONTAINER_NOT_FOUND);
+      }
+      candidate.writeLock();
+      Container<?> current = getContainer(containerId);
+      if (current == null) {
+        candidate.writeUnlock();
+        throw new StorageContainerException(
+            "Container " + containerId + " not found in ContainerSet.",
+            ContainerProtos.Result.CONTAINER_NOT_FOUND);
+      }
+      if (current != candidate) {
+        candidate.writeUnlock();
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Container {} mapping changed during lock acquisition (attempt {}); retrying.",
+              containerId, retry);
+        }
+        continue;
+      }
+      return candidate;
+    }
+    LOG.warn("Container {} mapping kept changing after {} attempts; giving up.",
+        containerId, MAX_CONTAINER_MAP_SWAP_RETRIES);
+    return null;
+  }
+
   /**
    * Removes container from both memory and database. This should be used when the containerData on disk has been
    * removed completely from the node.

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/DeleteBlocksCommandHandler.java

@@ -303,19 +303,31 @@ public DeleteBlockTransactionExecutionResult call() {
           if (keyValueContainer.
               writeLockTryLock(tryLockTimeoutMs, TimeUnit.MILLISECONDS)) {
             try {
-              String schemaVersion = containerData
-                  .getSupportedSchemaVersionOrDefault();
-              if (getSchemaHandlers().containsKey(schemaVersion)) {
-                schemaHandlers.get(schemaVersion).handle(containerData, tx);
+              // Re-fetch the container after acquiring the lock. DiskBalancer may have relocated
+              // this container to a different disk while we waited — in that case, the container
+              // object in ContainerSet has changed and containerData points to the old replica.
+              Container<?> current = containerSet.getContainer(containerId);
+              if (current == null || current.getContainerData() != containerData) {
+                LOG.debug("DeleteBlocks: containerData for container {} is stale "
+                    + ", Will retry on the new replica.",
+                    containerId);
+                lockAcquisitionFailed = true;
+                txResultBuilder.setContainerID(containerId).setSuccess(false);
               } else {
-                throw new UnsupportedOperationException(
-                    "Only schema version 1,2,3 are supported.");
+                String schemaVersion = containerData
+                    .getSupportedSchemaVersionOrDefault();
+                if (getSchemaHandlers().containsKey(schemaVersion)) {
+                  schemaHandlers.get(schemaVersion).handle(containerData, tx);
+                } else {
+                  throw new UnsupportedOperationException(
+                      "Only schema version 1,2,3 are supported.");
+                }
+                txResultBuilder.setContainerID(containerId)
+                    .setSuccess(true);
               }
             } finally {
               keyValueContainer.writeUnlock();
             }
-            txResultBuilder.setContainerID(containerId)
-                .setSuccess(true);
           } else {
             lockAcquisitionFailed = true;
             txResultBuilder.setContainerID(containerId)

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/diskbalancer/DiskBalancerService.java

@@ -512,19 +512,19 @@ public BackgroundTaskResult call() {
         return BackgroundTaskResult.EmptyTaskResult.newResult();
       }
 
-      // Double check container state before acquiring lock to start move process.
-      // Container state may have changed after selection.
-      State containerState = container.getContainerData().getState();
-      if (!movableContainerStates.contains(containerState)) {
-        LOG.warn("Container {} is in {} state, skipping move process.", containerId, containerState);
-        postCall(false, startTime);
-        return BackgroundTaskResult.EmptyTaskResult.newResult();
-      }
-
       // hold read lock on the container first, to avoid other threads to update the container state,
       // such as block deletion.
       container.readLock();
       try {
+        // Double check container state after acquiring lock to start move process.
+        // Container state may have changed after selection.
+        State containerState = container.getContainerData().getState();
+        if (!movableContainerStates.contains(containerState)) {
+          LOG.warn("Container {} is in {} state, skipping move process.", containerId, containerState);
+          moveSucceeded = false;
+          return BackgroundTaskResult.EmptyTaskResult.newResult();
+        }
+
         // Step 1: Copy container to new Volume's tmp Dir
         diskBalancerTmpDir = getDiskBalancerTmpDir(destVolume)
             .resolve(String.valueOf(containerId));
@@ -580,6 +580,10 @@ public BackgroundTaskResult call() {
         // old caller can still hold the old Container object.
         ozoneContainer.getContainerSet().updateContainer(newContainer);
         destVolume.incrementUsedSpace(containerSize);
+
+        // Test injector: ContainerSet now references newContainer while this thread still holds
+        // readLock on the old replica.
+        pauseInjector();
         // Mark old container as DELETED and persist state.
         // markContainerForDelete require writeLock, so release readLock first
         container.readUnlock();
@@ -597,13 +601,7 @@ public BackgroundTaskResult call() {
         metrics.incrSuccessBytes(containerSize);
         totalBalancedBytes.addAndGet(containerSize);
       } catch (IOException e) {
-        if (injector != null) {
-          try {
-            injector.pause();
-          } catch (IOException ex) {
-            // do nothing
-          }
-        }
+        pauseInjector();
         moveSucceeded = false;
         LOG.warn("Failed to move container {}", containerId, e);
         if (diskBalancerTmpDir != null) {
@@ -861,6 +859,17 @@ public static void setInjector(FaultInjector instance) {
     injector = instance;
   }
 
+  // call FaultInjector#pause when an injector is registered; ignore IOException.
+  private static void pauseInjector() {
+    if (injector != null) {
+      try {
+        injector.pause();
+      } catch (IOException ex) {
+        // do nothing
+      }
+    }
+  }
+
   @VisibleForTesting
   public void setReplicaDeletionDelay(long durationMills) {
     this.replicaDeletionDelay = durationMills;

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/diskbalancer/DiskBalancerVolumeCalculation.java

@@ -67,16 +67,41 @@ public static List<VolumeFixedUsage> getVolumeUsages(MutableVolumeSet volumeSet,
    * @param volumes Immutable list of volumes
    * from each source volume during container moves
    * @return Ideal usage as a ratio (used space / total capacity)
-   * @throws IllegalArgumentException if total capacity is zero
    */
   public static double getIdealUsage(List<VolumeFixedUsage> volumes) {
+    if (volumes == null || volumes.isEmpty()) {
+      return 0.0;
+    }
+
     long totalCapacity = 0L, totalEffectiveUsed = 0L;
-    
+
     for (VolumeFixedUsage volumeUsage : volumes) {
-      totalCapacity += volumeUsage.getUsage().getCapacity();
-      totalEffectiveUsed += volumeUsage.getEffectiveUsed();
+      final long capacity = volumeUsage.getUsage().getCapacity();
+      if (capacity < 0) {
+        throw new IllegalArgumentException(
+            "Negative capacity = " + capacity + ": " + volumeUsage.getVolume());
+      }
+
+      final long effectiveUsed = volumeUsage.getEffectiveUsed();
+      if (effectiveUsed < 0) {
+        throw new IllegalArgumentException(
+            "Negative effective used = " + effectiveUsed + ": "
+                + volumeUsage.getVolume());
+      }
+      if (effectiveUsed > capacity) {
+        throw new IllegalArgumentException(
+            "Effective used = " + effectiveUsed + " > capacity = "
+                + capacity + ": " + volumeUsage.getVolume());
+      }
+
+      totalCapacity += capacity;
+      totalEffectiveUsed += effectiveUsed;
     }
-    
+
+    if (totalCapacity == 0) {
+      return 0.0;
+    }
+
     return ((double) (totalEffectiveUsed)) / totalCapacity;
   }