diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationEndpoint.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationEndpoint.java index 3407189fcf6a..9c0494807c0a 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationEndpoint.java +++ b/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 entries; - int size; + long size; String walGroupId; int timeout; @@ -171,7 +171,7 @@ public ReplicateContext setEntries(List entries) { return this; } - public ReplicateContext setSize(int size) { + public ReplicateContext setSize(long size) { this.size = size; return this; } @@ -185,7 +185,7 @@ public List getEntries() { return entries; } - public int getSize() { + public long getSize() { return size; } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsSource.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsSource.java index 1c7a3fc84402..0c02d532baae 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/MetricsSource.java +++ b/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); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java index 5545a1fa0f50..2e121c622873 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java +++ b/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); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java index 69ad2887064a..cf3873dc4eda 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceInterface.java +++ b/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. diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java index deaa98e71ab9..f03817696464 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java +++ b/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 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()); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java index 9951bd709371..25e5011fdcde 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java @@ -362,14 +362,17 @@ private Pair 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 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 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. " diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationThrottler.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationThrottler.java index e39082c8e50a..abd090096273 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationThrottler.java +++ b/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; } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java index da0868be885f..c0fa5c1f0ba1 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/ReplicationSourceDummy.java +++ b/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 diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSizeOverflow.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSizeOverflow.java new file mode 100644 index 000000000000..d8e585a216a4 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSizeOverflow.java @@ -0,0 +1,121 @@ +/* + * 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> storeFiles = new HashMap<>(); + Map 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 >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 >2GB batch. + */ + @Test + public void testReplicateContextSizeDoesNotOverflow() { + ReplicationEndpoint.ReplicateContext ctx = new ReplicationEndpoint.ReplicateContext(); + ctx.setSize(OVER_2GB); + assertEquals(OVER_2GB, ctx.getSize()); + } +} diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationThrottler.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationThrottler.java index 078c3bc2d84b..5da4d76ffe4a 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationThrottler.java +++ b/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(); + } + } }