HBASE-30234: Replication shippedBytes metric overflows to negative due to int truncation of batch size

Several size-tracking variables in the replication source pipeline used int
instead of long, so once a WAL entry batch (or a bulk load's referenced store
files) exceeded Integer.MAX_VALUE (~2GB) the value silently overflowed. This
surfaced as a negative shippedBytes JMX metric and as broken throttling, since
the truncated (often negative) size was handed to the bandwidth throttler.

Widen the size to long end-to-end:
- ReplicationSourceShipper#shipEdits keeps getHeapSize() as long.
- ReplicationSourceInterface/ReplicationSource#tryThrottle and
  ReplicationThrottler#getNextSleepInterval accept a long size.
- MetricsSource#shipBatch takes a long sizeInBytes (sink counters are long).
- ReplicationEndpoint.ReplicateContext#size and its accessors are long.
- ReplicationSourceWALReader#sizeOfStoreFilesIncludeBulkLoad accumulates and
  returns long instead of casting each addition to int.

Add TestReplicationSizeOverflow (store-file-size accumulation, shippedBytes
metric path, ReplicateContext) and a >2GB case in TestReplicationThrottler.

Author: asdf2014

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationEndpoint.java

@@ -158,7 +158,7 @@ public Abortable getAbortable() {
   @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.REPLICATION)
   static class ReplicateContext {
     List<Entry> entries;
-    int size;
+    long size;
     String walGroupId;
     int timeout;
 
@@ -171,7 +171,7 @@ public ReplicateContext setEntries(List<Entry> entries) {
       return this;
     }
 
-    public ReplicateContext setSize(int size) {
+    public ReplicateContext setSize(long size) {
       this.size = size;
       return this;
     }
@@ -185,7 +185,7 @@ public List<Entry> getEntries() {
       return entries;
     }
 
-    public int getSize() {
+    public long getSize() {
       return size;
     }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsSource.java

@@ -210,7 +210,7 @@ public void incrLogEditsFiltered() {
    * Convience method to apply changes to metrics do to shipping a batch of logs.
    * @param batchSize the size of the batch that was shipped to sinks.
    */
-  public void shipBatch(long batchSize, int sizeInBytes) {
+  public void shipBatch(long batchSize, long sizeInBytes) {
     singleSourceSource.incrBatchesShipped(1);
     globalSourceSource.incrBatchesShipped(1);
 
@@ -258,7 +258,7 @@ public long getOpsShipped() {
    * @param batchSize the size of the batch that was shipped to sinks.
    * @param hfiles    total number of hfiles shipped to sinks.
    */
-  public void shipBatch(long batchSize, int sizeInBytes, long hfiles) {
+  public void shipBatch(long batchSize, long sizeInBytes, long hfiles) {
     shipBatch(batchSize, sizeInBytes);
     singleSourceSource.incrHFilesShipped(hfiles);
     globalSourceSource.incrHFilesShipped(hfiles);

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java

@@ -495,7 +495,7 @@ public ReplicationSourceManager getSourceManager() {
   }
 
   @Override
-  public void tryThrottle(int batchSize) throws InterruptedException {
+  public void tryThrottle(long batchSize) throws InterruptedException {
     checkBandwidthChangeAndResetThrottler();
     if (throttler.isEnabled()) {
       long sleepTicks = throttler.getNextSleepInterval(batchSize);

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java

@@ -167,7 +167,7 @@ default boolean isSyncReplication() {
    * Try to throttle when the peer config with a bandwidth
    * @param batchSize entries size will be pushed
    */
-  void tryThrottle(int batchSize) throws InterruptedException;
+  void tryThrottle(long batchSize) throws InterruptedException;
 
   /**
    * Call this after the shipper thread ship some entries to peer cluster.

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java

@@ -204,7 +204,7 @@ protected void postFinish() {
   void shipEdits(WALEntryBatch entryBatch) throws IOException {
     List<Entry> entries = entryBatch.getWalEntries();
     int sleepMultiplier = 0;
-    int currentSize = (int) entryBatch.getHeapSize();
+    long currentSize = entryBatch.getHeapSize();
     MetricsSource metrics = source.getSourceMetrics();
     if (metrics != null && !entries.isEmpty()) {
       metrics.setTimeStampNextToReplicate(entries.get(entries.size() - 1).getKey().getWriteTime());

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java

@@ -362,14 +362,17 @@ private Pair<Integer, Integer> countDistinctRowKeysAndHFiles(WALEdit edit) {
     return result;
   }
 
+  // Package-private static (a pure function of the edit) so it can be unit tested directly; see
+  // TestReplicationSizeOverflow. Uses long throughout to avoid overflow when the bulk load store
+  // files sum exceeds Integer.MAX_VALUE (~2GB).
   /**
    * Calculate the total size of all the store files
    * @param edit edit to count row keys from
    * @return the total size of the store files
    */
-  private int sizeOfStoreFilesIncludeBulkLoad(WALEdit edit) {
+  static long sizeOfStoreFilesIncludeBulkLoad(WALEdit edit) {
     List<Cell> cells = edit.getCells();
-    int totalStoreFilesSize = 0;
+    long totalStoreFilesSize = 0;
 
     int totalCells = edit.size();
     for (int i = 0; i < totalCells; i++) {
@@ -379,8 +382,7 @@ private int sizeOfStoreFilesIncludeBulkLoad(WALEdit edit) {
           List<StoreDescriptor> stores = bld.getStoresList();
           int totalStores = stores.size();
           for (int j = 0; j < totalStores; j++) {
-            totalStoreFilesSize =
-              (int) (totalStoreFilesSize + stores.get(j).getStoreFileSizeBytes());
+            totalStoreFilesSize += stores.get(j).getStoreFileSizeBytes();
           }
         } catch (IOException e) {
           LOG.error("Failed to deserialize bulk load entry from wal edit. "

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationThrottler.java

@@ -59,7 +59,7 @@ public boolean isEnabled() {
    * @param size is the size of edits to be pushed
    * @return sleep interval for throttling control
    */
-  public long getNextSleepInterval(final int size) {
+  public long getNextSleepInterval(final long size) {
     if (!this.enabled) {
       return 0;
     }

hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java

@@ -146,7 +146,7 @@ public ReplicationSourceManager getSourceManager() {
   }
 
   @Override
-  public void tryThrottle(int batchSize) throws InterruptedException {
+  public void tryThrottle(long batchSize) throws InterruptedException {
   }
 
   @Override

hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSizeOverflow.java

@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.replication.regionserver;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.RegionInfoBuilder;
+import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
+import org.apache.hadoop.hbase.testclassification.ReplicationTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.wal.WALEdit;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
+
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.BulkLoadDescriptor;
+
+/**
+ * Unit tests for HBASE-30234: several size-tracking variables in the replication source pipeline
+ * used {@code int} instead of {@code long}, causing integer overflow (and negative metrics /
+ * broken throttling) once a batch exceeds {@link Integer#MAX_VALUE} (~2GB). These tests exercise
+ * each fixed site with a value that overflows {@code int}.
+ */
+@Tag(ReplicationTests.TAG)
+@Tag(SmallTests.TAG)
+public class TestReplicationSizeOverflow {
+
+  private static final RegionInfo RI =
+    RegionInfoBuilder.newBuilder(TableName.valueOf("testReplicationSizeOverflow")).build();
+
+  /** A byte count that is larger than Integer.MAX_VALUE and would wrap negative as an int. */
+  private static final long OVER_2GB = 3_500_000_000L;
+
+  /**
+   * The sum of the bulk load store file sizes must not overflow when it exceeds ~2GB.
+   * {@link ReplicationSourceWALReader#sizeOfStoreFilesIncludeBulkLoad} previously accumulated the
+   * sizes into an int and cast on every step, wrapping the total to a negative value.
+   */
+  @Test
+  public void testSizeOfStoreFilesIncludeBulkLoadDoesNotOverflow() {
+    long size1 = 2_000_000_000L;
+    long size2 = 1_500_000_000L;
+    long expected = size1 + size2;
+    // sanity: the total genuinely overflows a signed int
+    assertTrue(expected > Integer.MAX_VALUE);
+
+    Map<byte[], List<Path>> storeFiles = new HashMap<>();
+    Map<String, Long> storeFilesSize = new HashMap<>();
+    Path hfile1 = new Path("f1");
+    storeFiles.put(Bytes.toBytes("f1"), Collections.singletonList(hfile1));
+    storeFilesSize.put(hfile1.getName(), size1);
+    Path hfile2 = new Path("f2");
+    storeFiles.put(Bytes.toBytes("f2"), Collections.singletonList(hfile2));
+    storeFilesSize.put(hfile2.getName(), size2);
+
+    BulkLoadDescriptor desc = ProtobufUtil.toBulkLoadDescriptor(RI.getTable(),
+      UnsafeByteOperations.unsafeWrap(RI.getEncodedNameAsBytes()), storeFiles, storeFilesSize, 1);
+    WALEdit edit = WALEdit.createBulkLoadEvent(RI, desc);
+
+    assertEquals(expected, ReplicationSourceWALReader.sizeOfStoreFilesIncludeBulkLoad(edit));
+  }
+
+  /**
+   * The shipped-bytes metric must receive the full batch size. Previously
+   * {@link MetricsSource#shipBatch} took an int {@code sizeInBytes}, so a &gt;2GB batch was
+   * incremented into the counter as a negative value.
+   */
+  @Test
+  public void testShipBatchDoesNotTruncateShippedBytes() {
+    MetricsReplicationSourceSource single = mock(MetricsReplicationSourceSource.class);
+    MetricsReplicationGlobalSourceSource global =
+      mock(MetricsReplicationGlobalSourceSource.class);
+    MetricsSource metrics = new MetricsSource("test-source", single, global, new HashMap<>());
+
+    metrics.shipBatch(10L, OVER_2GB, 2L);
+
+    // the full long size must reach both counters unchanged, not a truncated int
+    verify(single).incrShippedBytes(OVER_2GB);
+    verify(global).incrShippedBytes(OVER_2GB);
+    verify(single).incrHFilesShipped(2L);
+    verify(global).incrHFilesShipped(2L);
+  }
+
+  /**
+   * The replicate context must carry the full batch size to the endpoint. Previously the
+   * {@code size} field and its accessors were int, truncating a &gt;2GB batch.
+   */
+  @Test
+  public void testReplicateContextSizeDoesNotOverflow() {
+    ReplicationEndpoint.ReplicateContext ctx = new ReplicationEndpoint.ReplicateContext();
+    ctx.setSize(OVER_2GB);
+    assertEquals(OVER_2GB, ctx.getSize());
+  }
+}

hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationThrottler.java

@@ -22,6 +22,8 @@
 
 import org.apache.hadoop.hbase.testclassification.ReplicationTests;
 import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.ManualEnvironmentEdge;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
@@ -113,4 +115,33 @@ public void testThrottling() {
       assertTrue(ticks1 >= 375 && ticks1 <= 500);
     }
   }
+
+  /**
+   * HBASE-30234: a batch whose size exceeds Integer.MAX_VALUE (~2GB) must be treated as a large
+   * positive size instead of being silently truncated to a negative int. With truncation, the
+   * "delay to next cycle" branch would compare a negative sum against the bandwidth and wrongly
+   * return 0 (no throttling); with the fix it correctly delays the push to the next cycle.
+   */
+  @Test
+  public void testLargeSizeDoesNotOverflow() {
+    LOG.info("testLargeSizeDoesNotOverflow");
+    // Freeze the clock so the assertion is deterministic (no cycle boundary is crossed).
+    ManualEnvironmentEdge edge = new ManualEnvironmentEdge();
+    edge.setValue(1000);
+    EnvironmentEdgeManager.injectEdge(edge);
+    try {
+      // bandwidth of 1 byte/cycle so any positive push exceeds it
+      ReplicationThrottler throttler = new ReplicationThrottler(1);
+      // prime the current cycle so cyclePushSize > 0 (enables the "delay to next cycle" branch)
+      throttler.addPushSize(1);
+      // a size larger than Integer.MAX_VALUE; as an int this would wrap to a negative value
+      long hugeSize = (long) Integer.MAX_VALUE + 1L;
+      long ticks = throttler.getNextSleepInterval(hugeSize);
+      // delayed to next cycle: cycleStartTick(1000) + one cycle(100) - now(1000) == 100ms.
+      // A truncated (negative) size would fail the bandwidth check and wrongly return 0.
+      assertEquals(100, ticks);
+    } finally {
+      EnvironmentEdgeManager.reset();
+    }
+  }
 }