From 0e6818574098572fc6ac7c3a20c4f1575afd9a42 Mon Sep 17 00:00:00 2001 From: juke-mini666 Date: Fri, 19 Jun 2026 16:42:40 +0900 Subject: [PATCH 1/6] HBASE-30238 Add bandwidth throttling for bulkload HFile copy during replication Introduce hbase.replication.bulkload.copy.bandwidth.mb to rate-limit HFile copy from source HDFS in HFileReplicator.Copier. The limit is enforced through a shared RateLimiter across copy threads, and ReplicationSink updates the rate through configuration reload without requiring RegionServer restart. --- .../ReplicationSinkServiceImpl.java | 11 +- .../regionserver/HFileReplicator.java | 41 +++++- .../regionserver/ReplicationSink.java | 26 +++- .../TestHFileReplicatorBandwidth.java | 138 ++++++++++++++++++ 4 files changed, 207 insertions(+), 9 deletions(-) create mode 100644 hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHFileReplicatorBandwidth.java diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationSinkServiceImpl.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationSinkServiceImpl.java index acaf5756879f..5878a42d5f10 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationSinkServiceImpl.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationSinkServiceImpl.java @@ -28,6 +28,7 @@ import org.apache.hadoop.hbase.ScheduledChore; import org.apache.hadoop.hbase.Server; import org.apache.hadoop.hbase.Stoppable; +import org.apache.hadoop.hbase.conf.ConfigurationObserver; import org.apache.hadoop.hbase.regionserver.HRegionServer; import org.apache.hadoop.hbase.regionserver.RegionServerCoprocessorHost; import org.apache.hadoop.hbase.regionserver.ReplicationSinkService; @@ -41,7 +42,7 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos; @InterfaceAudience.Private -public class ReplicationSinkServiceImpl implements ReplicationSinkService { +public class ReplicationSinkServiceImpl implements ReplicationSinkService, ConfigurationObserver { private static final Logger LOG = LoggerFactory.getLogger(ReplicationSinkServiceImpl.class); private Configuration conf; @@ -90,6 +91,14 @@ public void stopReplicationService() { } } + @Override + public void onConfigurationChange(Configuration conf) { + this.conf = conf; + if (this.replicationSink != null) { + this.replicationSink.onConfigurationChange(conf); + } + } + @Override public ReplicationLoad refreshAndGetReplicationLoad() { if (replicationLoad == null) { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java index 2d0b4e32ced0..41ef6268e817 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java @@ -20,7 +20,9 @@ import java.io.Closeable; import java.io.FileNotFoundException; import java.io.IOException; +import java.io.InputStream; import java.io.InterruptedIOException; +import java.io.OutputStream; import java.math.BigInteger; import java.util.ArrayList; import java.util.Deque; @@ -37,7 +39,6 @@ import java.util.concurrent.TimeUnit; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.FileUtil; import org.apache.hadoop.fs.Path; import org.apache.hadoop.fs.permission.FsPermission; import org.apache.hadoop.hbase.HConstants; @@ -57,6 +58,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter; import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder; /** @@ -75,10 +77,20 @@ public class HFileReplicator implements Closeable { public static final String REPLICATION_BULKLOAD_COPY_HFILES_PERTHREAD_KEY = "hbase.replication.bulkload.copy.hfiles.perthread"; public static final int REPLICATION_BULKLOAD_COPY_HFILES_PERTHREAD_DEFAULT = 10; + /** + * Bandwidth limit in MB/s for copying HFiles from source cluster during bulkload replication. 0 + * means no limit. Can be changed dynamically via configuration reload. + */ + public static final String REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY = + "hbase.replication.bulkload.copy.bandwidth.mb"; + public static final double REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_DEFAULT = 0; private static final Logger LOG = LoggerFactory.getLogger(HFileReplicator.class); private static final String UNDERSCORE = "_"; private final static FsPermission PERM_ALL_ACCESS = FsPermission.valueOf("-rwxrwxrwx"); + private static final int COPY_BUFFER_SIZE = 65536; + // null means no throttling + private volatile RateLimiter rateLimiter; private Configuration sourceClusterConf; private String sourceBaseNamespaceDirPath; @@ -99,6 +111,14 @@ public HFileReplicator(Configuration sourceClusterConf, String sourceBaseNamespa String sourceHFileArchiveDirPath, Map>>> tableQueueMap, Configuration conf, AsyncClusterConnection connection, List sourceClusterIds) throws IOException { + this(sourceClusterConf, sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, tableQueueMap, + conf, connection, sourceClusterIds, RateLimiter.create(Double.MAX_VALUE)); + } + + public HFileReplicator(Configuration sourceClusterConf, String sourceBaseNamespaceDirPath, + String sourceHFileArchiveDirPath, Map>>> tableQueueMap, + Configuration conf, AsyncClusterConnection connection, List sourceClusterIds, + RateLimiter rateLimiter) throws IOException { this.sourceClusterConf = sourceClusterConf; this.sourceBaseNamespaceDirPath = sourceBaseNamespaceDirPath; this.sourceHFileArchiveDirPath = sourceHFileArchiveDirPath; @@ -120,6 +140,7 @@ public HFileReplicator(Configuration sourceClusterConf, String sourceBaseNamespa REPLICATION_BULKLOAD_COPY_HFILES_PERTHREAD_DEFAULT); sinkFs = FileSystem.get(conf); + this.rateLimiter = rateLimiter; } @Override @@ -336,16 +357,13 @@ public Void call() throws IOException { sourceHFilePath = new Path(sourceBaseNamespaceDirPath, hfiles.get(i)); localHFilePath = new Path(stagingDir, sourceHFilePath.getName()); try { - FileUtil.copy(sourceFs, sourceHFilePath, sinkFs, localHFilePath, false, conf); - // If any other exception other than FNFE then we will fail the replication requests and - // source will retry to replicate these data. + copyWithThrottle(sourceHFilePath, localHFilePath); } catch (FileNotFoundException e) { LOG.info("Failed to copy hfile from " + sourceHFilePath + " to " + localHFilePath + ". Trying to copy from hfile archive directory.", e); sourceHFilePath = new Path(sourceHFileArchiveDirPath, hfiles.get(i)); - try { - FileUtil.copy(sourceFs, sourceHFilePath, sinkFs, localHFilePath, false, conf); + copyWithThrottle(sourceHFilePath, localHFilePath); } catch (FileNotFoundException e1) { // This will mean that the hfile does not exists any where in source cluster FS. So we // cannot do anything here just log and continue. @@ -358,5 +376,16 @@ public Void call() throws IOException { } return null; } + + private void copyWithThrottle(Path src, Path dst) throws IOException { + try (InputStream in = sourceFs.open(src); OutputStream out = sinkFs.create(dst)) { + byte[] buf = new byte[COPY_BUFFER_SIZE]; + int bytesRead; + while ((bytesRead = in.read(buf)) >= 0) { + rateLimiter.acquire(bytesRead); + out.write(buf, 0, bytesRead); + } + } + } } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java index 508ace390565..f302141009b7 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java @@ -42,6 +42,7 @@ import java.util.stream.Collectors; import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.conf.ConfigurationObserver; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellUtil; @@ -71,6 +72,7 @@ import org.slf4j.LoggerFactory; import org.apache.hbase.thirdparty.com.google.common.collect.Lists; +import org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter; import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WALEntry; import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos; @@ -94,7 +96,7 @@ * TODO make this class more like ReplicationSource wrt log handling */ @InterfaceAudience.Private -public class ReplicationSink { +public class ReplicationSink implements ConfigurationObserver { private static final Logger LOG = LoggerFactory.getLogger(ReplicationSink.class); private final Configuration conf; @@ -108,6 +110,7 @@ public class ReplicationSink { private long hfilesReplicated = 0; private SourceFSConfigurationProvider provider; private WALEntrySinkFilter walEntrySinkFilter; + private final RateLimiter bulkLoadCopyRateLimiter = RateLimiter.create(Double.MAX_VALUE); /** * Row size threshold for multi requests above which a warning is logged @@ -143,6 +146,25 @@ public ReplicationSink(Configuration conf, RegionServerCoprocessorHost rsServerH throw new IllegalArgumentException( "Configured source fs configuration provider class " + className + " throws error.", e); } + updateBulkLoadCopyBandwidth(conf); + } + + @Override + public void onConfigurationChange(Configuration newConf) { + updateBulkLoadCopyBandwidth(newConf); + } + + double getBulkLoadCopyRateLimiterRate() { + return bulkLoadCopyRateLimiter.getRate(); + } + + private void updateBulkLoadCopyBandwidth(Configuration conf) { + double bandwidthMb = conf.getDouble(HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY, + HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_DEFAULT); + double newRate = bandwidthMb <= 0 ? Double.MAX_VALUE : bandwidthMb * 1024 * 1024; + bulkLoadCopyRateLimiter.setRate(newRate); + LOG.info("Bulkload copy bandwidth updated: {}", + bandwidthMb <= 0 ? "unlimited" : bandwidthMb + " MB/s"); } private WALEntrySinkFilter setupWALEntrySinkFilter() throws IOException { @@ -319,7 +341,7 @@ public void replicateEntries(List entries, final ExtendedCellScanner c Configuration providerConf = this.provider.getConf(this.conf, replicationClusterId); try (HFileReplicator hFileReplicator = new HFileReplicator(providerConf, sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, bulkLoadHFileMap, conf, - getConnection(), entry.getKey())) { + getConnection(), entry.getKey(), bulkLoadCopyRateLimiter)) { hFileReplicator.replicate(); LOG.debug("Finished replicating {} bulk loaded data", entry.getKey().toString()); } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHFileReplicatorBandwidth.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHFileReplicatorBandwidth.java new file mode 100644 index 000000000000..a30d7e56861e --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHFileReplicatorBandwidth.java @@ -0,0 +1,138 @@ +/* + * 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 java.io.IOException; +import java.io.OutputStream; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.HBaseTestingUtil; +import org.apache.hadoop.hbase.testclassification.ReplicationTests; +import org.apache.hadoop.hbase.testclassification.SmallTests; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +/** + * Unit tests for bulkload copy bandwidth throttling in {@link HFileReplicator} and + * {@link ReplicationSink}. Does not require a running HBase cluster. + */ +@Tag(ReplicationTests.TAG) +@Tag(SmallTests.TAG) +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class TestHFileReplicatorBandwidth { + + private final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); + + @BeforeAll + public void setUpBeforeClass() throws Exception { + TEST_UTIL.startMiniDFSCluster(1); + } + + @AfterAll + public void tearDownAfterClass() throws Exception { + TEST_UTIL.shutdownMiniDFSCluster(); + } + + /** + * Verify that onConfigurationChange updates the RateLimiter rate dynamically. + */ + @Test + public void testBandwidthDynamicUpdate() throws Exception { + Configuration conf = TEST_UTIL.getConfiguration(); + conf.set("hbase.replication.source.fs.conf.provider", + TestSourceFSConfigurationProvider.class.getCanonicalName()); + ReplicationSink sink = new ReplicationSink(conf, null); + + // Default: unlimited + assertEquals(Double.MAX_VALUE, sink.getBulkLoadCopyRateLimiterRate(), 0.0); + + // Apply 50 MB/s + Configuration newConf = new Configuration(conf); + newConf.setDouble(HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY, 50.0); + sink.onConfigurationChange(newConf); + assertEquals(50.0 * 1024 * 1024, sink.getBulkLoadCopyRateLimiterRate(), 1.0); + + // Reset to unlimited (0 = no limit) + Configuration resetConf = new Configuration(conf); + resetConf.setDouble(HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY, 0); + sink.onConfigurationChange(resetConf); + assertEquals(Double.MAX_VALUE, sink.getBulkLoadCopyRateLimiterRate(), 0.0); + } + + /** + * Verify that throttled copy actually slows down I/O at the specified rate. Creates a file of + * known size, copies it through a throttled stream at 1 MB/s, and asserts the elapsed time is at + * least (fileSize / 1MB * 0.8) seconds. + */ + @Test + public void testCopyIsThrottled() throws Exception { + Configuration conf = TEST_UTIL.getConfiguration(); + FileSystem fs = TEST_UTIL.getTestFileSystem(); + Path testDir = TEST_UTIL.getDataTestDirOnTestFS("testCopyIsThrottled"); + fs.mkdirs(testDir); + + // Write a ~2MB test file + Path srcFile = new Path(testDir, "src"); + int fileSize = 2 * 1024 * 1024; + try (FSDataOutputStream out = fs.create(srcFile)) { + writeBytes(out, fileSize); + } + + Path dstFile = new Path(testDir, "dst"); + + // Throttle at 1 MB/s + double limitMbPerSec = 1.0; + org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter limiter = + org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter + .create(limitMbPerSec * 1024 * 1024); + + long start = System.currentTimeMillis(); + try (java.io.InputStream in = fs.open(srcFile); OutputStream out = fs.create(dstFile)) { + byte[] buf = new byte[65536]; + int bytesRead; + while ((bytesRead = in.read(buf)) >= 0) { + limiter.acquire(bytesRead); + out.write(buf, 0, bytesRead); + } + } + long elapsed = System.currentTimeMillis() - start; + + // At 1MB/s, 2MB should take >= 1600ms (allow 20% margin for JVM overhead) + long minExpectedMs = (long) (fileSize * 1000L / (limitMbPerSec * 1024 * 1024) * 0.8); + assertTrue(elapsed >= minExpectedMs, + "Expected throttled copy >= " + minExpectedMs + "ms, got " + elapsed + "ms"); + } + + private static void writeBytes(OutputStream out, int size) throws IOException { + byte[] buf = new byte[65536]; + int written = 0; + while (written < size) { + int chunk = Math.min(buf.length, size - written); + out.write(buf, 0, chunk); + written += chunk; + } + } +} From 91c480f6e12aa2673af476002e99c8da4ce30179 Mon Sep 17 00:00:00 2001 From: juke-mini666 Date: Fri, 19 Jun 2026 16:42:59 +0900 Subject: [PATCH 2/6] HBASE-30238 Prevent duplicate bulkload replication on RPC retry Track in-progress bulkload events by replication cluster, encoded region, and bulkload sequence number so concurrent RPC retries skip duplicate execution. The key is removed after the current attempt completes or fails, preserving at-least-once retry semantics while avoiding concurrent duplicate loads. --- .../regionserver/ReplicationSink.java | 57 +++-- .../TestReplicationSinkBulkLoadDedup.java | 201 ++++++++++++++++++ 2 files changed, 246 insertions(+), 12 deletions(-) create mode 100644 hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java index f302141009b7..624d16e69491 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java @@ -35,14 +35,15 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Set; import java.util.TreeMap; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Collectors; import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hbase.conf.ConfigurationObserver; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellUtil; @@ -60,6 +61,7 @@ import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.RetriesExhaustedException; import org.apache.hadoop.hbase.client.Row; +import org.apache.hadoop.hbase.conf.ConfigurationObserver; import org.apache.hadoop.hbase.regionserver.RegionServerCoprocessorHost; import org.apache.hadoop.hbase.replication.ReplicationUtils; import org.apache.hadoop.hbase.security.UserProvider; @@ -111,6 +113,9 @@ public class ReplicationSink implements ConfigurationObserver { private SourceFSConfigurationProvider provider; private WALEntrySinkFilter walEntrySinkFilter; private final RateLimiter bulkLoadCopyRateLimiter = RateLimiter.create(Double.MAX_VALUE); + // Tracks bulkload events currently being processed to prevent duplicate execution on RPC retry. + // Key: replicationClusterId + "#" + encodedRegionName + "#" + bulkloadSeqNum + private final Set inProgressBulkLoads = ConcurrentHashMap.newKeySet(); /** * Row size threshold for multi requests above which a warning is logged @@ -158,6 +163,10 @@ public void onConfigurationChange(Configuration newConf) { return bulkLoadCopyRateLimiter.getRate(); } + Set getInProgressBulkLoads() { + return inProgressBulkLoads; + } + private void updateBulkLoadCopyBandwidth(Configuration conf) { double bandwidthMb = conf.getDouble(HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY, HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_DEFAULT); @@ -230,6 +239,8 @@ public void replicateEntries(List entries, final ExtendedCellScanner c Map, List>> rowMap = new TreeMap<>(); Map, Map>>>> bulkLoadsPerClusters = null; + // bulkload keys registered in inProgressBulkLoads for this batch, to be removed on completion + List registeredBulkLoadKeys = null; Pair, List> mutationsToWalEntriesPairs = new Pair<>(new ArrayList<>(), new ArrayList<>()); for (WALEntry entry : entries) { @@ -262,6 +273,16 @@ public void replicateEntries(List entries, final ExtendedCellScanner c if (CellUtil.matchingQualifier(cell, WALEdit.BULK_LOAD)) { BulkLoadDescriptor bld = WALEdit.getBulkLoadDescriptor(cell); if (bld.getReplicate()) { + String bulkLoadKey = buildBulkLoadKey(replicationClusterId, bld); + if (!inProgressBulkLoads.add(bulkLoadKey)) { + LOG.warn("Skipping duplicate bulkload replication, already in progress: {}", + bulkLoadKey); + continue; + } + if (registeredBulkLoadKeys == null) { + registeredBulkLoadKeys = new ArrayList<>(); + } + registeredBulkLoadKeys.add(bulkLoadKey); if (bulkLoadsPerClusters == null) { bulkLoadsPerClusters = new HashMap<>(); } @@ -333,19 +354,26 @@ public void replicateEntries(List entries, final ExtendedCellScanner c } if (bulkLoadsPerClusters != null) { - for (Entry, - Map>>>> entry : bulkLoadsPerClusters.entrySet()) { - Map>>> bulkLoadHFileMap = entry.getValue(); - if (bulkLoadHFileMap != null && !bulkLoadHFileMap.isEmpty()) { - LOG.debug("Replicating {} bulk loaded data", entry.getKey().toString()); - Configuration providerConf = this.provider.getConf(this.conf, replicationClusterId); - try (HFileReplicator hFileReplicator = new HFileReplicator(providerConf, - sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, bulkLoadHFileMap, conf, - getConnection(), entry.getKey(), bulkLoadCopyRateLimiter)) { - hFileReplicator.replicate(); - LOG.debug("Finished replicating {} bulk loaded data", entry.getKey().toString()); + try { + for (Entry, + Map>>>> entry : bulkLoadsPerClusters + .entrySet()) { + Map>>> bulkLoadHFileMap = entry.getValue(); + if (bulkLoadHFileMap != null && !bulkLoadHFileMap.isEmpty()) { + LOG.debug("Replicating {} bulk loaded data", entry.getKey().toString()); + Configuration providerConf = this.provider.getConf(this.conf, replicationClusterId); + try (HFileReplicator hFileReplicator = new HFileReplicator(providerConf, + sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, bulkLoadHFileMap, conf, + getConnection(), entry.getKey(), bulkLoadCopyRateLimiter)) { + hFileReplicator.replicate(); + LOG.debug("Finished replicating {} bulk loaded data", entry.getKey().toString()); + } } } + } finally { + if (registeredBulkLoadKeys != null) { + inProgressBulkLoads.removeAll(registeredBulkLoadKeys); + } } } @@ -444,6 +472,11 @@ private void addNewTableEntryInMap( bulkLoadHFileMap.put(tableName, newFamilyHFilePathsList); } + private static String buildBulkLoadKey(String replicationClusterId, BulkLoadDescriptor bld) { + return replicationClusterId + "#" + Bytes.toString(bld.getEncodedRegionName().toByteArray()) + + "#" + bld.getBulkloadSeqNum(); + } + private String getHFilePath(TableName table, BulkLoadDescriptor bld, String storeFile, byte[] family) { return new StringBuilder(100).append(table.getNamespaceAsString()).append(Path.SEPARATOR) diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java new file mode 100644 index 000000000000..dcb1b9c4d676 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java @@ -0,0 +1,201 @@ +/* + * 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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Collections; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseTestingUtil; +import org.apache.hadoop.hbase.HConstants; +import org.apache.hadoop.hbase.PrivateCellUtil; +import org.apache.hadoop.hbase.TableName; +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.apache.hadoop.hbase.wal.WALEditInternalHelper; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +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.AdminProtos.WALEntry; +import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.UUID; +import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos; +import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.WALKey; + +/** + * Unit tests for bulkload deduplication in {@link ReplicationSink}. Verifies that concurrent RPC + * retries for the same bulkload event do not result in duplicate processing. + */ +@Tag(ReplicationTests.TAG) +@Tag(SmallTests.TAG) +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class TestReplicationSinkBulkLoadDedup { + + private static final String CLUSTER_ID_A = "cluster-A"; + private static final String CLUSTER_ID_B = "cluster-B"; + private static final byte[] REGION_NAME = Bytes.toBytes("regionXYZ"); + private static final long SEQ_NUM = 100L; + private static final TableName TABLE = TableName.valueOf("testDedup"); + private static final byte[] FAMILY = Bytes.toBytes("cf"); + + private final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); + private ReplicationSink sink; + + @BeforeAll + public void setUpBeforeClass() throws Exception { + Configuration conf = TEST_UTIL.getConfiguration(); + conf.set("hbase.replication.source.fs.conf.provider", + TestSourceFSConfigurationProvider.class.getCanonicalName()); + sink = new ReplicationSink(conf, null); + } + + @AfterAll + public void tearDownAfterClass() { + // no cluster to shut down + } + + /** + * Verify that manually adding a key to inProgressBulkLoads blocks a second add for the same key, + * and that remove restores it — core Set semantics that the dedup logic relies on. + */ + @Test + public void testInProgressSetAddAndRemove() { + String key = CLUSTER_ID_A + "#" + Bytes.toString(REGION_NAME) + "#" + SEQ_NUM; + + assertTrue(sink.getInProgressBulkLoads().add(key), "First add should succeed"); + assertFalse(sink.getInProgressBulkLoads().add(key), + "Second add should fail (already in progress)"); + + sink.getInProgressBulkLoads().remove(key); + assertTrue(sink.getInProgressBulkLoads().add(key), + "Add after remove should succeed (retry allowed)"); + sink.getInProgressBulkLoads().remove(key); // cleanup + } + + /** + * Verify that keys from different source clusters with the same region/seqNum are treated as + * distinct entries and do not block each other. + */ + @Test + public void testDifferentClustersDontConflict() { + String keyA = CLUSTER_ID_A + "#" + Bytes.toString(REGION_NAME) + "#" + SEQ_NUM; + String keyB = CLUSTER_ID_B + "#" + Bytes.toString(REGION_NAME) + "#" + SEQ_NUM; + + assertTrue(sink.getInProgressBulkLoads().add(keyA)); + assertTrue(sink.getInProgressBulkLoads().add(keyB), + "Same region/seqNum from different cluster should not conflict"); + + sink.getInProgressBulkLoads().remove(keyA); + sink.getInProgressBulkLoads().remove(keyB); + assertEquals(0, sink.getInProgressBulkLoads().size()); + } + + /** + * End-to-end: replicateEntries() with a bulkload WAL cell that has replicate=false should not + * register any key in inProgressBulkLoads. + */ + @Test + public void testNonReplicateBulkLoadNotTracked() throws Exception { + WALProtos.BulkLoadDescriptor bld = buildBulkLoadDescriptor(REGION_NAME, SEQ_NUM, false); + WALEdit edit = buildWALEdit(bld); + List entries = buildWALEntries(edit); + + int before = sink.getInProgressBulkLoads().size(); + sink.replicateEntries(entries, + PrivateCellUtil + .createExtendedCellScanner(WALEditInternalHelper.getExtendedCells(edit).iterator()), + CLUSTER_ID_A, "/dummy/namespace", "/dummy/archive"); + + assertEquals(before, sink.getInProgressBulkLoads().size(), + "Non-replicate bulkload should not add key to inProgressBulkLoads"); + } + + /** + * Simulate a concurrent retry: manually pre-populate the key to mimic a first call still in + * progress, then verify replicateEntries() skips the bulkload cell without throwing. + */ + @Test + public void testConcurrentRetryIsSkipped() throws Exception { + WALProtos.BulkLoadDescriptor bld = buildBulkLoadDescriptor(REGION_NAME, SEQ_NUM + 1, true); + String key = CLUSTER_ID_A + "#" + Bytes.toString(REGION_NAME) + "#" + (SEQ_NUM + 1); + + // Simulate first call still in progress + sink.getInProgressBulkLoads().add(key); + + WALEdit edit = buildWALEdit(bld); + List entries = buildWALEntries(edit); + + // Second call (retry) should skip without exception + sink.replicateEntries(entries, + PrivateCellUtil + .createExtendedCellScanner(WALEditInternalHelper.getExtendedCells(edit).iterator()), + CLUSTER_ID_A, "/dummy/namespace", "/dummy/archive"); + + // Key still held by the "first call" + assertTrue(sink.getInProgressBulkLoads().contains(key)); + sink.getInProgressBulkLoads().remove(key); // cleanup + } + + // ---- helpers ---- + + private WALProtos.BulkLoadDescriptor buildBulkLoadDescriptor(byte[] regionName, long seqNum, + boolean replicate) { + WALProtos.StoreDescriptor store = + WALProtos.StoreDescriptor.newBuilder().setFamilyName(UnsafeByteOperations.unsafeWrap(FAMILY)) + .setStoreHomeDir(Bytes.toString(FAMILY)).addStoreFile("hfile-0").setStoreFileSizeBytes(1024) + .build(); + return WALProtos.BulkLoadDescriptor.newBuilder() + .setTableName(ProtobufUtil.toProtoTableName(TABLE)) + .setEncodedRegionName(UnsafeByteOperations.unsafeWrap(regionName)).addStores(store) + .setBulkloadSeqNum(seqNum).setReplicate(replicate).build(); + } + + private WALEdit buildWALEdit(WALProtos.BulkLoadDescriptor bld) { + // RegionInfo is only used to construct the WAL cell row key; dedup logic reads + // encodedRegionName from BulkLoadDescriptor directly, so any RegionInfo works here. + org.apache.hadoop.hbase.client.RegionInfo ri = + org.apache.hadoop.hbase.client.RegionInfoBuilder.newBuilder(TABLE).build(); + return WALEdit.createBulkLoadEvent(ri, bld); + } + + private List buildWALEntries(WALEdit edit) { + WALEntry.Builder builder = WALEntry.newBuilder(); + builder.setAssociatedCellCount(edit.getCells().size()); + WALKey.Builder keyBuilder = WALKey.newBuilder(); + UUID.Builder uuidBuilder = UUID.newBuilder(); + uuidBuilder.setLeastSigBits(HConstants.DEFAULT_CLUSTER_ID.getLeastSignificantBits()); + uuidBuilder.setMostSigBits(HConstants.DEFAULT_CLUSTER_ID.getMostSignificantBits()); + keyBuilder.setClusterId(uuidBuilder.build()); + keyBuilder.setTableName(UnsafeByteOperations.unsafeWrap(TABLE.getName())); + keyBuilder.setWriteTime(System.currentTimeMillis()); + keyBuilder.setEncodedRegionName(UnsafeByteOperations.unsafeWrap(REGION_NAME)); + keyBuilder.setLogSequenceNumber(-1); + builder.setKey(keyBuilder.build()); + return Collections.singletonList(builder.build()); + } +} From 99d9281efb79b8129bc297edd4e46ca66d0dbb0a Mon Sep 17 00:00:00 2001 From: juke-mini666 Date: Sun, 21 Jun 2026 09:23:37 +0900 Subject: [PATCH 3/6] HBASE-30238 Address bulkload replication review comments --- .../hbase/replication/regionserver/HFileReplicator.java | 4 +++- .../hbase/replication/regionserver/ReplicationSink.java | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java index 41ef6268e817..1cfa56bb6d8f 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java @@ -79,7 +79,9 @@ public class HFileReplicator implements Closeable { public static final int REPLICATION_BULKLOAD_COPY_HFILES_PERTHREAD_DEFAULT = 10; /** * Bandwidth limit in MB/s for copying HFiles from source cluster during bulkload replication. 0 - * means no limit. Can be changed dynamically via configuration reload. + * means no limit. Can be changed dynamically via configuration reload. A low limit can make the + * sink-side bulkload copy exceed the replication RPC timeout, so configure the timeout + * accordingly. */ public static final String REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY = "hbase.replication.bulkload.copy.bandwidth.mb"; diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java index 624d16e69491..e979ae009e85 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java @@ -472,7 +472,7 @@ private void addNewTableEntryInMap( bulkLoadHFileMap.put(tableName, newFamilyHFilePathsList); } - private static String buildBulkLoadKey(String replicationClusterId, BulkLoadDescriptor bld) { + private String buildBulkLoadKey(String replicationClusterId, BulkLoadDescriptor bld) { return replicationClusterId + "#" + Bytes.toString(bld.getEncodedRegionName().toByteArray()) + "#" + bld.getBulkloadSeqNum(); } From a54186fea140a473cfb6d711b0fcab0219f078e6 Mon Sep 17 00:00:00 2001 From: juke-mini666 Date: Sun, 21 Jun 2026 16:37:22 +0900 Subject: [PATCH 4/6] HBASE-30238 Add distributed bulkload replication event tracking Coordinate replicated bulkload WAL events across target region servers using ZooKeeper in-progress and completed markers. Add master-side cleanup for completed markers and cover cross-RS retry replay with MiniCluster. --- .../apache/hadoop/hbase/master/HMaster.java | 6 + .../ReplicationBulkLoadEventCleaner.java | 70 +++++ .../ReplicationSinkServiceImpl.java | 2 +- .../ReplicationBulkLoadEventTracker.java | 268 ++++++++++++++++++ .../regionserver/ReplicationSink.java | 141 +++++++-- .../TestReplicationBulkLoadEventCleaner.java | 149 ++++++++++ .../replication/TestBulkLoadReplication.java | 179 ++++++++++++ .../TestReplicationBulkLoadEventTracker.java | 113 ++++++++ .../TestReplicationSinkBulkLoadDedup.java | 37 ++- 9 files changed, 930 insertions(+), 35 deletions(-) create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBulkLoadEventCleaner.java create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationBulkLoadEventTracker.java create mode 100644 hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBulkLoadEventCleaner.java create mode 100644 hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationBulkLoadEventTracker.java diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java index 0be892584354..e6095db8688d 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java @@ -141,6 +141,7 @@ import org.apache.hadoop.hbase.master.cleaner.HFileCleaner; import org.apache.hadoop.hbase.master.cleaner.LogCleaner; import org.apache.hadoop.hbase.master.cleaner.ReplicationBarrierCleaner; +import org.apache.hadoop.hbase.master.cleaner.ReplicationBulkLoadEventCleaner; import org.apache.hadoop.hbase.master.cleaner.SnapshotCleanerChore; import org.apache.hadoop.hbase.master.hbck.HbckChore; import org.apache.hadoop.hbase.master.http.MasterDumpServlet; @@ -435,6 +436,7 @@ public class HMaster extends HBaseServerBase implements Maste // The exclusive hfile cleaner pool for scanning the archive directory private DirScanPool exclusiveHFileCleanerPool; private ReplicationBarrierCleaner replicationBarrierCleaner; + private ReplicationBulkLoadEventCleaner replicationBulkLoadEventCleaner; private MobFileCleanerChore mobFileCleanerChore; private MobFileCompactionChore mobFileCompactionChore; private RollingUpgradeChore rollingUpgradeChore; @@ -1803,6 +1805,9 @@ conf, getMasterFileSystem().getFileSystem(), new Path(archiveDir, path), replicationBarrierCleaner = new ReplicationBarrierCleaner(conf, this, getConnection(), replicationPeerManager); getChoreService().scheduleChore(replicationBarrierCleaner); + replicationBulkLoadEventCleaner = + new ReplicationBulkLoadEventCleaner(conf, this, getZooKeeper()); + getChoreService().scheduleChore(replicationBulkLoadEventCleaner); final boolean isSnapshotChoreEnabled = this.snapshotCleanupStateStore.get(); this.snapshotCleanerChore = new SnapshotCleanerChore(this, conf, getSnapshotManager()); @@ -1990,6 +1995,7 @@ protected void stopChores() { hfileCleaners = null; } shutdownChore(replicationBarrierCleaner); + shutdownChore(replicationBulkLoadEventCleaner); shutdownChore(snapshotCleanerChore); shutdownChore(hbckChore); shutdownChore(regionsRecoveryChore); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBulkLoadEventCleaner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBulkLoadEventCleaner.java new file mode 100644 index 000000000000..d0491c01fadd --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBulkLoadEventCleaner.java @@ -0,0 +1,70 @@ +/* + * 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.master.cleaner; + +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.ScheduledChore; +import org.apache.hadoop.hbase.Stoppable; +import org.apache.hadoop.hbase.replication.regionserver.ReplicationBulkLoadEventTracker; +import org.apache.hadoop.hbase.zookeeper.ZKWatcher; +import org.apache.yetus.audience.InterfaceAudience; +import org.apache.zookeeper.KeeperException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Cleans completed replicated bulk load event markers after they are old enough. + */ +@InterfaceAudience.Private +public class ReplicationBulkLoadEventCleaner extends ScheduledChore { + + private static final Logger LOG = LoggerFactory.getLogger(ReplicationBulkLoadEventCleaner.class); + + public static final String PERIOD_MS_KEY = + "hbase.master.cleaner.replication.bulkload.event.period.ms"; + public static final int PERIOD_MS_DEFAULT = (int) TimeUnit.MINUTES.toMillis(10); + public static final String DONE_TTL_MS_KEY = "hbase.replication.bulkload.event.done.ttl.ms"; + public static final long DONE_TTL_MS_DEFAULT = TimeUnit.DAYS.toMillis(1); + + private final ReplicationBulkLoadEventTracker tracker; + private final long doneTtlMs; + + public ReplicationBulkLoadEventCleaner(Configuration conf, Stoppable stopper, ZKWatcher zkw) { + super("ReplicationBulkLoadEventCleaner", stopper, + conf.getInt(PERIOD_MS_KEY, PERIOD_MS_DEFAULT)); + this.tracker = new ReplicationBulkLoadEventTracker(conf, zkw); + this.doneTtlMs = conf.getLong(DONE_TTL_MS_KEY, DONE_TTL_MS_DEFAULT); + } + + @Override + protected void chore() { + try { + int deleted = tracker.cleanDoneMarkers(doneTtlMs); + if (deleted > 0) { + LOG.info("Cleaned {} replicated bulkload event marker(s)", deleted); + } + } catch (KeeperException e) { + LOG.warn("Failed to clean replicated bulkload event markers", e); + } + } + + public void choreForTesting() { + chore(); + } +} diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationSinkServiceImpl.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationSinkServiceImpl.java index 5878a42d5f10..7ce677297e88 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationSinkServiceImpl.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ReplicationSinkServiceImpl.java @@ -79,7 +79,7 @@ public void startReplicationService() throws IOException { if (server instanceof HRegionServer) { rsServerHost = ((HRegionServer) server).getRegionServerCoprocessorHost(); } - this.replicationSink = new ReplicationSink(this.conf, rsServerHost); + this.replicationSink = new ReplicationSink(this.conf, rsServerHost, server.getZooKeeper()); this.server.getChoreService().scheduleChore(new ReplicationStatisticsChore( "ReplicationSinkStatistics", server, (int) TimeUnit.SECONDS.toMillis(statsPeriodInSecond))); } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationBulkLoadEventTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationBulkLoadEventTracker.java new file mode 100644 index 000000000000..a4d0153552dd --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationBulkLoadEventTracker.java @@ -0,0 +1,268 @@ +/* + * 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 java.io.IOException; +import java.io.InterruptedIOException; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; +import org.apache.hadoop.hbase.util.MD5Hash; +import org.apache.hadoop.hbase.zookeeper.ZKUtil; +import org.apache.hadoop.hbase.zookeeper.ZKWatcher; +import org.apache.hadoop.hbase.zookeeper.ZNodePaths; +import org.apache.yetus.audience.InterfaceAudience; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.data.Stat; + +/** + * Coordinates replicated bulk load events across all region servers in the sink cluster. + */ +@InterfaceAudience.Private +public class ReplicationBulkLoadEventTracker { + + public static final String ZNODE_KEY = "hbase.replication.bulkload.event.tracker.znode"; + public static final String ZNODE_DEFAULT = "bulkload-events"; + public static final String BUCKET_WIDTH_MS_KEY = + "hbase.replication.bulkload.event.bucket.width.ms"; + public static final long BUCKET_WIDTH_MS_DEFAULT = TimeUnit.HOURS.toMillis(1); + public static final String WAIT_TIMEOUT_MS_KEY = + "hbase.replication.bulkload.event.wait.timeout.ms"; + public static final long WAIT_TIMEOUT_MS_DEFAULT = TimeUnit.MINUTES.toMillis(1); + public static final String WAIT_INTERVAL_MS_KEY = + "hbase.replication.bulkload.event.wait.interval.ms"; + public static final long WAIT_INTERVAL_MS_DEFAULT = TimeUnit.SECONDS.toMillis(1); + + private static final byte[] EMPTY_BYTES = new byte[0]; + private static final String IN_PROGRESS = "in-progress"; + private static final String DONE = "done"; + + private final ZKWatcher zkw; + private final long bucketWidthMs; + private final long waitTimeoutMs; + private final long waitIntervalMs; + private final String inProgressZNode; + private final String doneZNode; + + public ReplicationBulkLoadEventTracker(Configuration conf, ZKWatcher zkw) { + this.zkw = Objects.requireNonNull(zkw, "zkw"); + this.bucketWidthMs = Math.max(1, conf.getLong(BUCKET_WIDTH_MS_KEY, BUCKET_WIDTH_MS_DEFAULT)); + this.waitTimeoutMs = Math.max(0, conf.getLong(WAIT_TIMEOUT_MS_KEY, WAIT_TIMEOUT_MS_DEFAULT)); + this.waitIntervalMs = Math.max(1, conf.getLong(WAIT_INTERVAL_MS_KEY, WAIT_INTERVAL_MS_DEFAULT)); + + String baseZNode = ZNodePaths.joinZNode(this.zkw.getZNodePaths().replicationZNode, + conf.get(ZNODE_KEY, ZNODE_DEFAULT)); + this.inProgressZNode = ZNodePaths.joinZNode(baseZNode, IN_PROGRESS); + this.doneZNode = ZNodePaths.joinZNode(baseZNode, DONE); + } + + public Event newEvent(String replicationClusterId, TableName table, byte[] encodedRegionName, + long bulkLoadSeqNum, long writeTime) { + String bucket = Long.toString(Math.max(0, writeTime) / bucketWidthMs); + String eventKey = replicationClusterId + '\n' + table.getNameWithNamespaceInclAsString() + '\n' + + Bytes.toStringBinary(encodedRegionName) + '\n' + bulkLoadSeqNum; + return new Event(bucket, MD5Hash.getMD5AsHex(Bytes.toBytes(eventKey)), eventKey); + } + + public ClaimResult claim(Event event) throws IOException { + long deadline = EnvironmentEdgeManager.currentTime() + waitTimeoutMs; + String inProgressPath = getInProgressPath(event); + try { + while (true) { + if (isDone(event)) { + return ClaimResult.COMPLETED; + } + ZKUtil.createWithParents(zkw, ZKUtil.getParent(inProgressPath)); + if (ZKUtil.createEphemeralNodeAndWatch(zkw, inProgressPath, event.getData())) { + return ClaimResult.CLAIMED; + } + if (isDone(event)) { + return ClaimResult.COMPLETED; + } + long now = EnvironmentEdgeManager.currentTime(); + if (now >= deadline) { + throw new IOException("Timed out waiting for replicated bulkload event " + event); + } + sleep(Math.min(waitIntervalMs, deadline - now)); + } + } catch (KeeperException e) { + throw new IOException("Failed to claim replicated bulkload event " + event, e); + } + } + + public void markDone(Event event) throws IOException { + try { + ZKUtil.createSetData(zkw, getDonePath(event), event.getData()); + } catch (KeeperException e) { + throw new IOException("Failed to mark replicated bulkload event done " + event, e); + } + } + + public void release(Event event) throws IOException { + try { + ZKUtil.deleteNodeFailSilent(zkw, getInProgressPath(event)); + } catch (KeeperException e) { + throw new IOException("Failed to release replicated bulkload event " + event, e); + } + } + + public boolean isInProgress(Event event) throws KeeperException { + return ZKUtil.checkExists(zkw, getInProgressPath(event)) != -1; + } + + public boolean isDone(Event event) throws KeeperException { + return ZKUtil.checkExists(zkw, getDonePath(event)) != -1; + } + + public int cleanDoneMarkers(long ttlMs) throws KeeperException { + long minAgeMs = Math.max(0, ttlMs); + long now = EnvironmentEdgeManager.currentTime(); + List buckets = ZKUtil.listChildrenNoWatch(zkw, doneZNode); + if (buckets == null) { + return 0; + } + int deleted = 0; + for (String bucket : buckets) { + String doneBucketPath = ZNodePaths.joinZNode(doneZNode, bucket); + List eventIds = ZKUtil.listChildrenNoWatch(zkw, doneBucketPath); + if (eventIds == null) { + continue; + } + for (String eventId : eventIds) { + String donePath = ZNodePaths.joinZNode(doneBucketPath, eventId); + Stat stat = new Stat(); + if (ZKUtil.getDataNoWatch(zkw, donePath, stat) == null) { + continue; + } + if (now - stat.getMtime() < minAgeMs) { + continue; + } + String inProgressPath = + ZNodePaths.joinZNode(ZNodePaths.joinZNode(inProgressZNode, bucket), eventId); + if (ZKUtil.checkExists(zkw, inProgressPath) != -1) { + continue; + } + ZKUtil.deleteNodeFailSilent(zkw, donePath); + deleted++; + } + deleteIfEmpty(doneBucketPath); + deleteIfEmpty(ZNodePaths.joinZNode(inProgressZNode, bucket)); + } + return deleted; + } + + private void deleteIfEmpty(String znode) throws KeeperException { + List children = ZKUtil.listChildrenNoWatch(zkw, znode); + if (children == null || !children.isEmpty()) { + return; + } + try { + ZKUtil.deleteNodeFailSilent(zkw, znode); + } catch (KeeperException.NotEmptyException e) { + // Another region server added a child after our list; keep the bucket. + } + } + + private String getInProgressPath(Event event) { + return ZNodePaths.joinZNode(ZNodePaths.joinZNode(inProgressZNode, event.getBucket()), + event.getEventId()); + } + + private String getDonePath(Event event) { + return ZNodePaths.joinZNode(ZNodePaths.joinZNode(doneZNode, event.getBucket()), + event.getEventId()); + } + + private void sleep(long sleepMs) throws InterruptedIOException { + try { + Thread.sleep(Math.max(1, sleepMs)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + InterruptedIOException ioe = + new InterruptedIOException("Interrupted while waiting for replicated bulkload event"); + ioe.initCause(e); + throw ioe; + } + } + + public enum ClaimResult { + CLAIMED(true), + COMPLETED(false); + + private final boolean claimed; + + ClaimResult(boolean claimed) { + this.claimed = claimed; + } + + public boolean isClaimed() { + return claimed; + } + } + + public static final class Event { + private final String bucket; + private final String eventId; + private final String eventKey; + + private Event(String bucket, String eventId, String eventKey) { + this.bucket = bucket; + this.eventId = eventId; + this.eventKey = eventKey; + } + + public String getBucket() { + return bucket; + } + + public String getEventId() { + return eventId; + } + + private byte[] getData() { + return eventKey == null ? EMPTY_BYTES : eventKey.getBytes(StandardCharsets.UTF_8); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof Event)) { + return false; + } + Event event = (Event) o; + return Objects.equals(bucket, event.bucket) && Objects.equals(eventId, event.eventId); + } + + @Override + public int hashCode() { + return Objects.hash(bucket, eventId); + } + + @Override + public String toString() { + return "Event{bucket='" + bucket + "', eventId='" + eventId + "'}"; + } + } +} diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java index e979ae009e85..c2c984809d10 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java @@ -69,6 +69,7 @@ import org.apache.hadoop.hbase.util.FutureUtils; import org.apache.hadoop.hbase.util.Pair; import org.apache.hadoop.hbase.wal.WALEdit; +import org.apache.hadoop.hbase.zookeeper.ZKWatcher; import org.apache.yetus.audience.InterfaceAudience; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -116,6 +117,7 @@ public class ReplicationSink implements ConfigurationObserver { // Tracks bulkload events currently being processed to prevent duplicate execution on RPC retry. // Key: replicationClusterId + "#" + encodedRegionName + "#" + bulkloadSeqNum private final Set inProgressBulkLoads = ConcurrentHashMap.newKeySet(); + private final ReplicationBulkLoadEventTracker bulkLoadEventTracker; /** * Row size threshold for multi requests above which a warning is logged @@ -132,8 +134,15 @@ public class ReplicationSink implements ConfigurationObserver { */ public ReplicationSink(Configuration conf, RegionServerCoprocessorHost rsServerHost) throws IOException { + this(conf, rsServerHost, null); + } + + public ReplicationSink(Configuration conf, RegionServerCoprocessorHost rsServerHost, + ZKWatcher zkw) throws IOException { this.conf = HBaseConfiguration.create(conf); this.rsServerHost = rsServerHost; + this.bulkLoadEventTracker = + zkw == null ? null : new ReplicationBulkLoadEventTracker(this.conf, zkw); rowSizeWarnThreshold = conf.getInt(HConstants.BATCH_ROWS_THRESHOLD_NAME, HConstants.BATCH_ROWS_THRESHOLD_DEFAULT); replicationSinkTrackerEnabled = conf.getBoolean(REPLICATION_SINK_TRACKER_ENABLED_KEY, @@ -230,6 +239,9 @@ public void replicateEntries(List entries, final ExtendedCellScanner c if (entries.isEmpty()) { return; } + // Bulkload keys/events registered for this batch, to be removed on completion or failure. + List registeredBulkLoadKeys = null; + List claimedBulkLoadEvents = null; // Very simple optimization where we batch sequences of rows going // to the same table. try { @@ -239,8 +251,8 @@ public void replicateEntries(List entries, final ExtendedCellScanner c Map, List>> rowMap = new TreeMap<>(); Map, Map>>>> bulkLoadsPerClusters = null; - // bulkload keys registered in inProgressBulkLoads for this batch, to be removed on completion - List registeredBulkLoadKeys = null; + Map, List> bulkLoadEventsPerClusters = + null; Pair, List> mutationsToWalEntriesPairs = new Pair<>(new ArrayList<>(), new ArrayList<>()); for (WALEntry entry : entries) { @@ -273,24 +285,50 @@ public void replicateEntries(List entries, final ExtendedCellScanner c if (CellUtil.matchingQualifier(cell, WALEdit.BULK_LOAD)) { BulkLoadDescriptor bld = WALEdit.getBulkLoadDescriptor(cell); if (bld.getReplicate()) { - String bulkLoadKey = buildBulkLoadKey(replicationClusterId, bld); - if (!inProgressBulkLoads.add(bulkLoadKey)) { - LOG.warn("Skipping duplicate bulkload replication, already in progress: {}", - bulkLoadKey); - continue; + ReplicationBulkLoadEventTracker.Event bulkLoadEvent = null; + if (bulkLoadEventTracker != null) { + bulkLoadEvent = bulkLoadEventTracker.newEvent(replicationClusterId, table, + bld.getEncodedRegionName().toByteArray(), bld.getBulkloadSeqNum(), + entry.getKey().getWriteTime()); + ReplicationBulkLoadEventTracker.ClaimResult claimResult = + bulkLoadEventTracker.claim(bulkLoadEvent); + if (!claimResult.isClaimed()) { + LOG.info("Skipping completed bulkload replication event {}", bulkLoadEvent); + continue; + } + if (claimedBulkLoadEvents == null) { + claimedBulkLoadEvents = new ArrayList<>(); + } + claimedBulkLoadEvents.add(bulkLoadEvent); + } else { + String bulkLoadKey = buildBulkLoadKey(replicationClusterId, bld); + if (!inProgressBulkLoads.add(bulkLoadKey)) { + LOG.warn("Skipping duplicate bulkload replication, already in progress: {}", + bulkLoadKey); + continue; + } + if (registeredBulkLoadKeys == null) { + registeredBulkLoadKeys = new ArrayList<>(); + } + registeredBulkLoadKeys.add(bulkLoadKey); } - if (registeredBulkLoadKeys == null) { - registeredBulkLoadKeys = new ArrayList<>(); - } - registeredBulkLoadKeys.add(bulkLoadKey); if (bulkLoadsPerClusters == null) { bulkLoadsPerClusters = new HashMap<>(); } // Map of table name Vs list of pair of family and list of // hfile paths from its namespace + List clusterIds = entry.getKey().getClusterIdsList().stream() + .map(k -> toUUID(k).toString()).collect(Collectors.toList()); Map>>> bulkLoadHFileMap = - bulkLoadsPerClusters.computeIfAbsent(bld.getClusterIdsList(), k -> new HashMap<>()); + bulkLoadsPerClusters.computeIfAbsent(clusterIds, k -> new HashMap<>()); buildBulkLoadHFileMap(bulkLoadHFileMap, table, bld); + if (bulkLoadEvent != null) { + if (bulkLoadEventsPerClusters == null) { + bulkLoadEventsPerClusters = new HashMap<>(); + } + bulkLoadEventsPerClusters.computeIfAbsent(clusterIds, k -> new ArrayList<>()) + .add(bulkLoadEvent); + } } } else if (CellUtil.matchingQualifier(cell, WALEdit.REPLICATION_MARKER)) { Mutation put = processReplicationMarkerEntry(cell); @@ -354,26 +392,22 @@ public void replicateEntries(List entries, final ExtendedCellScanner c } if (bulkLoadsPerClusters != null) { - try { - for (Entry, - Map>>>> entry : bulkLoadsPerClusters - .entrySet()) { - Map>>> bulkLoadHFileMap = entry.getValue(); - if (bulkLoadHFileMap != null && !bulkLoadHFileMap.isEmpty()) { - LOG.debug("Replicating {} bulk loaded data", entry.getKey().toString()); - Configuration providerConf = this.provider.getConf(this.conf, replicationClusterId); - try (HFileReplicator hFileReplicator = new HFileReplicator(providerConf, - sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, bulkLoadHFileMap, conf, - getConnection(), entry.getKey(), bulkLoadCopyRateLimiter)) { - hFileReplicator.replicate(); - LOG.debug("Finished replicating {} bulk loaded data", entry.getKey().toString()); - } + for (Entry, + Map>>>> entry : bulkLoadsPerClusters.entrySet()) { + Map>>> bulkLoadHFileMap = entry.getValue(); + if (bulkLoadHFileMap != null && !bulkLoadHFileMap.isEmpty()) { + LOG.debug("Replicating {} bulk loaded data", entry.getKey().toString()); + Configuration providerConf = this.provider.getConf(this.conf, replicationClusterId); + try (HFileReplicator hFileReplicator = new HFileReplicator(providerConf, + sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, bulkLoadHFileMap, conf, + getConnection(), entry.getKey(), bulkLoadCopyRateLimiter)) { + hFileReplicator.replicate(); + markBulkLoadEventsDone(bulkLoadEventsPerClusters == null + ? null + : bulkLoadEventsPerClusters.get(entry.getKey()), claimedBulkLoadEvents); + LOG.debug("Finished replicating {} bulk loaded data", entry.getKey().toString()); } } - } finally { - if (registeredBulkLoadKeys != null) { - inProgressBulkLoads.removeAll(registeredBulkLoadKeys); - } } } @@ -385,6 +419,53 @@ public void replicateEntries(List entries, final ExtendedCellScanner c LOG.error("Unable to accept edit because:", ex); this.metrics.incrementFailedBatches(); throw ex; + } finally { + if (registeredBulkLoadKeys != null) { + inProgressBulkLoads.removeAll(registeredBulkLoadKeys); + } + releaseBulkLoadEvents(claimedBulkLoadEvents); + } + } + + private void markBulkLoadEventsDone(List events, + List releasableEvents) throws IOException { + if (bulkLoadEventTracker == null || events == null) { + return; + } + for (int i = 0; i < events.size(); i++) { + ReplicationBulkLoadEventTracker.Event event = events.get(i); + try { + bulkLoadEventTracker.markDone(event); + } catch (IOException e) { + // HFiles may already be loaded, so keep unmarked events in-progress while this RS session + // is alive. Releasing them here could let another sink repeat the same bulkload. + keepUnmarkedBulkLoadEventsInProgress(events, releasableEvents, i); + throw e; + } + } + } + + private void keepUnmarkedBulkLoadEventsInProgress( + List events, + List releasableEvents, int firstUnmarkedIndex) { + if (releasableEvents == null) { + return; + } + for (int i = firstUnmarkedIndex; i < events.size(); i++) { + releasableEvents.remove(events.get(i)); + } + } + + private void releaseBulkLoadEvents(List events) { + if (bulkLoadEventTracker == null || events == null) { + return; + } + for (ReplicationBulkLoadEventTracker.Event event : events) { + try { + bulkLoadEventTracker.release(event); + } catch (IOException e) { + LOG.warn("Failed to release replicated bulkload event {}", event, e); + } } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBulkLoadEventCleaner.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBulkLoadEventCleaner.java new file mode 100644 index 000000000000..1282d467e1c9 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBulkLoadEventCleaner.java @@ -0,0 +1,149 @@ +/* + * 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.master.cleaner; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseTestingUtil; +import org.apache.hadoop.hbase.Stoppable; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.replication.regionserver.ReplicationBulkLoadEventTracker; +import org.apache.hadoop.hbase.testclassification.MediumTests; +import org.apache.hadoop.hbase.testclassification.ReplicationTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.zookeeper.ZKUtil; +import org.apache.hadoop.hbase.zookeeper.ZKWatcher; +import org.apache.hadoop.hbase.zookeeper.ZNodePaths; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +@Tag(ReplicationTests.TAG) +@Tag(MediumTests.TAG) +public class TestReplicationBulkLoadEventCleaner { + + private static final HBaseTestingUtil UTIL = new HBaseTestingUtil(); + private static final TableName TABLE = TableName.valueOf("testBulkLoadEventCleaner"); + private static final byte[] REGION_NAME = Bytes.toBytes("region-1"); + private static final long WRITE_TIME = 123456789L; + private static final long SEQ_NUM = 100L; + + private static ZKWatcher zkw; + + @BeforeAll + public static void setUpBeforeClass() throws Exception { + UTIL.startMiniZKCluster(); + zkw = new ZKWatcher(UTIL.getConfiguration(), "bulkload-event-cleaner", null); + } + + @AfterAll + public static void tearDownAfterClass() throws Exception { + if (zkw != null) { + zkw.close(); + } + UTIL.shutdownMiniZKCluster(); + } + + @Test + public void testCleanerRemovesExpiredDoneWithoutInProgress() throws Exception { + Configuration conf = UTIL.getConfiguration(); + conf.setLong(ReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); + conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 1); + + ReplicationBulkLoadEventTracker tracker = new ReplicationBulkLoadEventTracker(conf, zkw); + ReplicationBulkLoadEventTracker.Event event = + tracker.newEvent("cluster-A", TABLE, REGION_NAME, SEQ_NUM, WRITE_TIME); + tracker.markDone(event); + assertTrue(tracker.isDone(event)); + + Thread.sleep(5); + ReplicationBulkLoadEventCleaner cleaner = + new ReplicationBulkLoadEventCleaner(conf, new NeverStopped(), zkw); + cleaner.choreForTesting(); + + assertFalse(tracker.isDone(event)); + } + + @Test + public void testCleanerKeepsDoneWhileInProgressExists() throws Exception { + Configuration conf = UTIL.getConfiguration(); + conf.setLong(ReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); + conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 1); + + ReplicationBulkLoadEventTracker tracker = new ReplicationBulkLoadEventTracker(conf, zkw); + ReplicationBulkLoadEventTracker.Event event = + tracker.newEvent("cluster-A", TABLE, REGION_NAME, SEQ_NUM + 1, WRITE_TIME); + assertTrue(tracker.claim(event).isClaimed()); + tracker.markDone(event); + + Thread.sleep(5); + ReplicationBulkLoadEventCleaner cleaner = + new ReplicationBulkLoadEventCleaner(conf, new NeverStopped(), zkw); + cleaner.choreForTesting(); + + assertTrue(tracker.isDone(event)); + tracker.release(event); + } + + @Test + public void testCleanerRemovesEmptyBucketsAfterExpiredDone() throws Exception { + Configuration conf = UTIL.getConfiguration(); + conf.setLong(ReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); + conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 1); + + ReplicationBulkLoadEventTracker tracker = new ReplicationBulkLoadEventTracker(conf, zkw); + ReplicationBulkLoadEventTracker.Event event = + tracker.newEvent("cluster-A", TABLE, REGION_NAME, SEQ_NUM + 2, WRITE_TIME); + assertTrue(tracker.claim(event).isClaimed()); + tracker.markDone(event); + tracker.release(event); + + String eventsRoot = ZNodePaths.joinZNode(zkw.getZNodePaths().replicationZNode, + ReplicationBulkLoadEventTracker.ZNODE_DEFAULT); + String doneBucketPath = + ZNodePaths.joinZNode(ZNodePaths.joinZNode(eventsRoot, "done"), event.getBucket()); + String inProgressBucketPath = + ZNodePaths.joinZNode(ZNodePaths.joinZNode(eventsRoot, "in-progress"), event.getBucket()); + + assertTrue(ZKUtil.checkExists(zkw, doneBucketPath) != -1); + assertTrue(ZKUtil.checkExists(zkw, inProgressBucketPath) != -1); + + Thread.sleep(5); + ReplicationBulkLoadEventCleaner cleaner = + new ReplicationBulkLoadEventCleaner(conf, new NeverStopped(), zkw); + cleaner.choreForTesting(); + + assertEquals(-1, ZKUtil.checkExists(zkw, doneBucketPath)); + assertEquals(-1, ZKUtil.checkExists(zkw, inProgressBucketPath)); + } + + private static final class NeverStopped implements Stoppable { + @Override + public void stop(String why) { + } + + @Override + public boolean isStopped() { + return false; + } + } +} diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestBulkLoadReplication.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestBulkLoadReplication.java index cac9e8e68260..8d151491c73b 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestBulkLoadReplication.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestBulkLoadReplication.java @@ -28,6 +28,7 @@ import java.io.IOException; import java.net.UnknownHostException; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -37,6 +38,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellBuilderType; @@ -45,12 +47,15 @@ import org.apache.hadoop.hbase.HBaseTestingUtil; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.PrivateCellUtil; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.Admin; import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.RegionInfo; +import org.apache.hadoop.hbase.client.RegionLocator; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.client.TableDescriptor; @@ -62,11 +67,19 @@ import org.apache.hadoop.hbase.io.hfile.HFile; import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder; import org.apache.hadoop.hbase.regionserver.HRegion; +import org.apache.hadoop.hbase.regionserver.RegionServerCoprocessorHost; +import org.apache.hadoop.hbase.replication.regionserver.ReplicationSink; import org.apache.hadoop.hbase.testclassification.LargeTests; import org.apache.hadoop.hbase.testclassification.ReplicationTests; import org.apache.hadoop.hbase.tool.BulkLoadHFilesTool; import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.CommonFSUtils; +import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; +import org.apache.hadoop.hbase.util.JVMClusterUtil; import org.apache.hadoop.hbase.util.Pair; +import org.apache.hadoop.hbase.wal.WALEdit; +import org.apache.hadoop.hbase.wal.WALEditInternalHelper; +import org.apache.hadoop.hbase.zookeeper.ZKWatcher; import org.apache.hadoop.hdfs.MiniDFSCluster; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterEach; @@ -77,6 +90,14 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +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.AdminProtos.WALEntry; +import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.UUID; +import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos; +import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.WALKey; + /** * Integration test for bulk load replication. Defines three clusters, with the following * replication topology: "1 <-> 2 <-> 3" (active-active between 1 and 2, and active-active between 2 @@ -235,6 +256,164 @@ public void testBulkLoadReplicationActiveActive() throws Exception { assertEquals(9, BULK_LOADS_COUNT.get()); } + @Test + public void testDuplicateBulkLoadWalEventSkippedAcrossTargetRegionServers() throws Exception { + byte[] row = Bytes.toBytes("duplicate-rs-retry"); + byte[] value = Bytes.toBytes("dedup-value"); + BulkLoadWalEvent bulkLoadWalEvent = createBulkLoadWalEvent(UTIL1, row, value); + ReplicationSink firstSink = null; + ReplicationSink retrySink = null; + JVMClusterUtil.RegionServerThread retryRegionServer = null; + boolean peer1Disabled = false; + boolean peer3Disabled = false; + try { + // Keep this test focused on the direct UTIL1 -> UTIL2 replay path. The fixture is + // active-active, so UTIL2's outgoing peers would otherwise increment the global observer. + peer1Disabled = disableReplicationPeerIfEnabled(UTIL2, PEER_ID1); + peer3Disabled = disableReplicationPeerIfEnabled(UTIL2, PEER_ID3); + firstSink = newReplicationSink(UTIL2, 0); + BULK_LOAD_LATCH = new CountDownLatch(1); + // The first sink simulates the original replication RPC and performs the actual bulk load. + firstSink.replicateEntries(bulkLoadWalEvent.entries, + PrivateCellUtil.createExtendedCellScanner( + WALEditInternalHelper.getExtendedCells(bulkLoadWalEvent.edit).iterator()), + PEER1_CLUSTER_ID, bulkLoadWalEvent.sourceBaseNamespaceDir, + bulkLoadWalEvent.sourceHFileArchiveDir); + + assertTrue(BULK_LOAD_LATCH.await(1, TimeUnit.MINUTES)); + assertTableHasValue(htable2, row, value); + assertEquals(1, BULK_LOADS_COUNT.get()); + + retryRegionServer = UTIL2.getMiniHBaseCluster().startRegionServer(); + setupCoprocessor(UTIL2); + retrySink = new ReplicationSink(CONF2, + retryRegionServer.getRegionServer().getRegionServerCoprocessorHost(), + retryRegionServer.getRegionServer().getZooKeeper()); + + // A source-side retry may choose a different target RS sink, so replay the exact same WAL + // batch through another ReplicationSink and verify that the completed marker skips it. + retrySink.replicateEntries(bulkLoadWalEvent.entries, + PrivateCellUtil.createExtendedCellScanner( + WALEditInternalHelper.getExtendedCells(bulkLoadWalEvent.edit).iterator()), + PEER1_CLUSTER_ID, bulkLoadWalEvent.sourceBaseNamespaceDir, + bulkLoadWalEvent.sourceHFileArchiveDir); + + assertEquals(1, BULK_LOADS_COUNT.get()); + } finally { + if (retrySink != null) { + retrySink.stopReplicationSinkServices(); + } + if (firstSink != null) { + firstSink.stopReplicationSinkServices(); + } + if (retryRegionServer != null) { + UTIL2.getMiniHBaseCluster() + .stopRegionServer(retryRegionServer.getRegionServer().getServerName()); + UTIL2.getMiniHBaseCluster().waitForRegionServerToStop( + retryRegionServer.getRegionServer().getServerName(), TimeUnit.MINUTES.toMillis(1)); + } + if (peer3Disabled) { + UTIL2.getAdmin().enableReplicationPeer(PEER_ID3); + } + if (peer1Disabled) { + UTIL2.getAdmin().enableReplicationPeer(PEER_ID1); + } + } + } + + private boolean disableReplicationPeerIfEnabled(HBaseTestingUtil utility, String peerId) + throws IOException { + boolean enabled = utility.getAdmin().listReplicationPeers().stream() + .anyMatch(peer -> peerId.equals(peer.getPeerId()) && peer.isEnabled()); + if (enabled) { + utility.getAdmin().disableReplicationPeer(peerId); + } + return enabled; + } + + private ReplicationSink newReplicationSink(HBaseTestingUtil utility, int regionServerIndex) + throws IOException { + RegionServerCoprocessorHost rsCpHost = utility.getMiniHBaseCluster() + .getRegionServer(regionServerIndex).getRegionServerCoprocessorHost(); + ZKWatcher zkw = utility.getMiniHBaseCluster().getRegionServer(regionServerIndex).getZooKeeper(); + return new ReplicationSink(utility.getConfiguration(), rsCpHost, zkw); + } + + private BulkLoadWalEvent createBulkLoadWalEvent(HBaseTestingUtil source, byte[] row, byte[] value) + throws Exception { + String bulkLoadFilePath = createHFileForFamilies(row, value, source.getConfiguration()); + Path sourceBaseNamespaceDir = + new Path(CommonFSUtils.getRootDir(source.getConfiguration()), HConstants.BASE_NAMESPACE_DIR); + Path sourceHFileArchiveDir = new Path(CommonFSUtils.getRootDir(source.getConfiguration()), + new Path(HConstants.HFILE_ARCHIVE_DIRECTORY, HConstants.BASE_NAMESPACE_DIR)); + try (RegionLocator locator = source.getConnection().getRegionLocator(tableName)) { + RegionInfo regionInfo = locator.getRegionLocation(row).getRegion(); + Path sourceHFilePath = copyBulkLoadFileToSourceCluster(source, bulkLoadFilePath, + sourceBaseNamespaceDir, regionInfo.getEncodedName(), Bytes.toString(famName)); + WALEdit edit = createBulkLoadWALEdit(regionInfo, sourceHFilePath, source.getConfiguration()); + WALEntry entry = createBulkLoadWALEntry(tableName, regionInfo, edit.getCells().size()); + return new BulkLoadWalEvent(Collections.singletonList(entry), edit, + sourceBaseNamespaceDir.toString(), sourceHFileArchiveDir.toString()); + } + } + + private Path copyBulkLoadFileToSourceCluster(HBaseTestingUtil source, String bulkLoadFilePath, + Path sourceBaseNamespaceDir, String encodedRegionName, String familyName) throws IOException { + Path sourceHFilePath = new Path(sourceBaseNamespaceDir, + tableName.getNamespaceAsString() + Path.SEPARATOR + tableName.getQualifierAsString() + + Path.SEPARATOR + encodedRegionName + Path.SEPARATOR + familyName + Path.SEPARATOR + + new Path(bulkLoadFilePath).getName()); + FileSystem sourceFs = source.getDFSCluster().getFileSystem(); + sourceFs.mkdirs(sourceHFilePath.getParent()); + sourceFs.copyFromLocalFile(new Path(bulkLoadFilePath), sourceHFilePath); + return sourceHFilePath; + } + + private WALEdit createBulkLoadWALEdit(RegionInfo regionInfo, Path sourceHFilePath, + Configuration sourceConf) throws IOException { + Map> storeFiles = new HashMap<>(1); + storeFiles.put(famName, Collections.singletonList(sourceHFilePath)); + Map storeFilesSize = new HashMap<>(1); + storeFilesSize.put(sourceHFilePath.getName(), + sourceHFilePath.getFileSystem(sourceConf).getFileStatus(sourceHFilePath).getLen()); + long bulkLoadSeqNum = EnvironmentEdgeManager.currentTime(); + WALProtos.BulkLoadDescriptor loadDescriptor = ProtobufUtil.toBulkLoadDescriptor(tableName, + UnsafeByteOperations.unsafeWrap(regionInfo.getEncodedNameAsBytes()), storeFiles, + storeFilesSize, bulkLoadSeqNum); + return WALEdit.createBulkLoadEvent(regionInfo, loadDescriptor); + } + + private WALEntry createBulkLoadWALEntry(TableName tableName, RegionInfo regionInfo, + int associatedCellCount) { + UUID.Builder uuidBuilder = UUID.newBuilder(); + uuidBuilder.setLeastSigBits(HConstants.DEFAULT_CLUSTER_ID.getLeastSignificantBits()); + uuidBuilder.setMostSigBits(HConstants.DEFAULT_CLUSTER_ID.getMostSignificantBits()); + WALKey.Builder keyBuilder = WALKey.newBuilder(); + keyBuilder.setClusterId(uuidBuilder.build()); + keyBuilder.setTableName(UnsafeByteOperations.unsafeWrap(tableName.getName())); + keyBuilder.setWriteTime(EnvironmentEdgeManager.currentTime()); + keyBuilder + .setEncodedRegionName(UnsafeByteOperations.unsafeWrap(regionInfo.getEncodedNameAsBytes())); + keyBuilder.setLogSequenceNumber(1); + return WALEntry.newBuilder().setAssociatedCellCount(associatedCellCount) + .setKey(keyBuilder.build()).build(); + } + + private static final class BulkLoadWalEvent { + private final List entries; + private final WALEdit edit; + private final String sourceBaseNamespaceDir; + private final String sourceHFileArchiveDir; + + private BulkLoadWalEvent(List entries, WALEdit edit, String sourceBaseNamespaceDir, + String sourceHFileArchiveDir) { + this.entries = entries; + this.edit = edit; + this.sourceBaseNamespaceDir = sourceBaseNamespaceDir; + this.sourceHFileArchiveDir = sourceHFileArchiveDir; + } + } + protected void assertBulkLoadConditions(TableName tableName, byte[] row, byte[] value, HBaseTestingUtil utility, Table... tables) throws Exception { BULK_LOAD_LATCH = new CountDownLatch(3); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationBulkLoadEventTracker.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationBulkLoadEventTracker.java new file mode 100644 index 000000000000..11d7dde34680 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationBulkLoadEventTracker.java @@ -0,0 +1,113 @@ +/* + * 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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseTestingUtil; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.testclassification.MediumTests; +import org.apache.hadoop.hbase.testclassification.ReplicationTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.zookeeper.ZKWatcher; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +@Tag(ReplicationTests.TAG) +@Tag(MediumTests.TAG) +public class TestReplicationBulkLoadEventTracker { + + private static final HBaseTestingUtil UTIL = new HBaseTestingUtil(); + private static final TableName TABLE = TableName.valueOf("testBulkLoadEventTracker"); + private static final byte[] REGION_NAME = Bytes.toBytes("region-1"); + private static final long WRITE_TIME = 123456789L; + private static final long SEQ_NUM = 100L; + + private static ZKWatcher zkw1; + private static ZKWatcher zkw2; + + @BeforeAll + public static void setUpBeforeClass() throws Exception { + UTIL.startMiniZKCluster(); + zkw1 = new ZKWatcher(UTIL.getConfiguration(), "tracker-1", null); + zkw2 = new ZKWatcher(UTIL.getConfiguration(), "tracker-2", null); + } + + @AfterAll + public static void tearDownAfterClass() throws Exception { + if (zkw2 != null) { + zkw2.close(); + } + if (zkw1 != null) { + zkw1.close(); + } + UTIL.shutdownMiniZKCluster(); + } + + @Test + public void testSecondSinkWaitsForDoneFromFirstSink() throws Exception { + Configuration conf = UTIL.getConfiguration(); + conf.setLong(ReplicationBulkLoadEventTracker.WAIT_TIMEOUT_MS_KEY, 5000); + conf.setLong(ReplicationBulkLoadEventTracker.WAIT_INTERVAL_MS_KEY, 50); + + ReplicationBulkLoadEventTracker tracker1 = new ReplicationBulkLoadEventTracker(conf, zkw1); + ReplicationBulkLoadEventTracker tracker2 = new ReplicationBulkLoadEventTracker(conf, zkw2); + ReplicationBulkLoadEventTracker.Event event = + tracker1.newEvent("cluster-A", TABLE, REGION_NAME, SEQ_NUM, WRITE_TIME); + + assertEquals(ReplicationBulkLoadEventTracker.ClaimResult.CLAIMED, tracker1.claim(event)); + assertTrue(tracker1.isInProgress(event)); + assertFalse(tracker2.isDone(event)); + + Thread marker = new Thread(() -> { + try { + Thread.sleep(200); + tracker1.markDone(event); + tracker1.release(event); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + marker.start(); + assertEquals(ReplicationBulkLoadEventTracker.ClaimResult.COMPLETED, tracker2.claim(event)); + marker.join(); + + assertFalse(tracker1.isInProgress(event)); + assertTrue(tracker2.isDone(event)); + } + + @Test + public void testDoneMarkerUsesStableBucketFromWalWriteTime() throws Exception { + Configuration conf = UTIL.getConfiguration(); + conf.setLong(ReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); + ReplicationBulkLoadEventTracker tracker = new ReplicationBulkLoadEventTracker(conf, zkw1); + + ReplicationBulkLoadEventTracker.Event first = + tracker.newEvent("cluster-A", TABLE, REGION_NAME, SEQ_NUM, WRITE_TIME); + ReplicationBulkLoadEventTracker.Event retry = + tracker.newEvent("cluster-A", TABLE, REGION_NAME, SEQ_NUM, WRITE_TIME); + + assertEquals(first.getBucket(), retry.getBucket()); + assertEquals(first.getEventId(), retry.getEventId()); + } +} diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java index dcb1b9c4d676..13cf39406006 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java @@ -28,11 +28,12 @@ import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.PrivateCellUtil; import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.testclassification.MediumTests; 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.apache.hadoop.hbase.wal.WALEditInternalHelper; +import org.apache.hadoop.hbase.zookeeper.ZKWatcher; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Tag; @@ -52,7 +53,7 @@ * retries for the same bulkload event do not result in duplicate processing. */ @Tag(ReplicationTests.TAG) -@Tag(SmallTests.TAG) +@Tag(MediumTests.TAG) @TestInstance(TestInstance.Lifecycle.PER_CLASS) public class TestReplicationSinkBulkLoadDedup { @@ -65,18 +66,24 @@ public class TestReplicationSinkBulkLoadDedup { private final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); private ReplicationSink sink; + private ZKWatcher zkw; @BeforeAll public void setUpBeforeClass() throws Exception { Configuration conf = TEST_UTIL.getConfiguration(); conf.set("hbase.replication.source.fs.conf.provider", TestSourceFSConfigurationProvider.class.getCanonicalName()); + TEST_UTIL.startMiniZKCluster(); + zkw = new ZKWatcher(conf, "replication-sink-bulkload-dedup", null); sink = new ReplicationSink(conf, null); } @AfterAll - public void tearDownAfterClass() { - // no cluster to shut down + public void tearDownAfterClass() throws Exception { + if (zkw != null) { + zkw.close(); + } + TEST_UTIL.shutdownMiniZKCluster(); } /** @@ -161,6 +168,28 @@ public void testConcurrentRetryIsSkipped() throws Exception { sink.getInProgressBulkLoads().remove(key); // cleanup } + @Test + public void testCompletedZkBulkLoadEventIsSkippedBySink() throws Exception { + WALProtos.BulkLoadDescriptor bld = buildBulkLoadDescriptor(REGION_NAME, SEQ_NUM + 2, true); + WALEdit edit = buildWALEdit(bld); + List entries = buildWALEntries(edit); + + ReplicationBulkLoadEventTracker tracker = + new ReplicationBulkLoadEventTracker(TEST_UTIL.getConfiguration(), zkw); + ReplicationBulkLoadEventTracker.Event event = tracker.newEvent(CLUSTER_ID_A, TABLE, REGION_NAME, + SEQ_NUM + 2, entries.get(0).getKey().getWriteTime()); + tracker.markDone(event); + + ReplicationSink zkSink = new ReplicationSink(TEST_UTIL.getConfiguration(), null, zkw); + zkSink.replicateEntries(entries, + PrivateCellUtil + .createExtendedCellScanner(WALEditInternalHelper.getExtendedCells(edit).iterator()), + CLUSTER_ID_A, "/missing/namespace", "/missing/archive"); + + assertTrue(tracker.isDone(event)); + assertFalse(tracker.isInProgress(event)); + } + // ---- helpers ---- private WALProtos.BulkLoadDescriptor buildBulkLoadDescriptor(byte[] regionName, long seqNum, From 364a78568a8f425c7b2c4c67b6374de3f56f8d9b Mon Sep 17 00:00:00 2001 From: juke-mini666 Date: Tue, 23 Jun 2026 08:05:30 +0900 Subject: [PATCH 5/6] HBASE-30238 Preserve bulkload descriptor cluster ids for replication Keep using BulkLoadDescriptor cluster ids when invoking HFileReplicator so replicated bulkload RPCs do not treat the target cluster as already handled. The ZooKeeper event tracking identity remains unchanged. --- .../hadoop/hbase/replication/regionserver/ReplicationSink.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java index c2c984809d10..6a73b714a71e 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java @@ -317,8 +317,7 @@ public void replicateEntries(List entries, final ExtendedCellScanner c } // Map of table name Vs list of pair of family and list of // hfile paths from its namespace - List clusterIds = entry.getKey().getClusterIdsList().stream() - .map(k -> toUUID(k).toString()).collect(Collectors.toList()); + List clusterIds = bld.getClusterIdsList(); Map>>> bulkLoadHFileMap = bulkLoadsPerClusters.computeIfAbsent(clusterIds, k -> new HashMap<>()); buildBulkLoadHFileMap(bulkLoadHFileMap, table, bld); From acab6f7b09d04e15ba5d96986239fc2532009c7f Mon Sep 17 00:00:00 2001 From: juke-mini666 Date: Thu, 16 Jul 2026 15:11:11 +0900 Subject: [PATCH 6/6] HBASE-30238 Prevent duplicate replicated bulk loads across retries Split replicated bulkload event tracking behind an interface and keep the ZooKeeper implementation responsible for marker lifecycle. Mark loaded table events done immediately and keep failed table events in progress to avoid retrying already loaded HFiles. --- .../src/main/resources/hbase-default.xml | 73 +++++ .../ReplicationBulkLoadEventCleaner.java | 11 +- .../regionserver/HFileReplicator.java | 30 +- .../ReplicationBulkLoadEventTracker.java | 193 ++----------- .../regionserver/ReplicationSink.java | 86 ++++-- .../ZKReplicationBulkLoadEventTracker.java | 259 ++++++++++++++++++ .../TestReplicationBulkLoadEventCleaner.java | 45 ++- .../TestHFileReplicatorBandwidth.java | 90 +----- .../TestReplicationBulkLoadEventTracker.java | 12 +- .../TestReplicationSinkBulkLoadDedup.java | 250 +++++++++++------ 10 files changed, 649 insertions(+), 400 deletions(-) create mode 100644 hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ZKReplicationBulkLoadEventTracker.java diff --git a/hbase-common/src/main/resources/hbase-default.xml b/hbase-common/src/main/resources/hbase-default.xml index e921fd499427..b81a13be6bb1 100644 --- a/hbase-common/src/main/resources/hbase-default.xml +++ b/hbase-common/src/main/resources/hbase-default.xml @@ -1867,6 +1867,79 @@ possible configurations would overwhelm and obscure the important. default of 10 will rarely need to be changed. + + hbase.replication.bulkload.copy.bandwidth.mb + 0 + + The maximum aggregate bandwidth, in megabytes per second, that a sink RegionServer uses while + copying HFiles from the source cluster for replicated bulk loads. 0 means no limit. This value + can be changed dynamically through configuration reload on the sink. If this is set too low, + the HFile copy can exceed the replication RPC timeout and cause the source to retry the same + batch. + + + + hbase.replication.bulkload.event.tracker.znode + bulkload-events + + The ZooKeeper znode name under the HBase replication znode used to coordinate replicated + bulk load events in the sink cluster. RegionServers create short-lived in-progress markers + and completed-event markers under this znode so that retries of the same replicated bulk + load WAL event are not executed more than once across different target RegionServers. + + + + hbase.replication.bulkload.event.bucket.width.ms + 3600000 + + The time width, in milliseconds, of each bucket used when storing replicated bulk load event + markers. The event write time is divided by this value to choose a bucket. Larger values + reduce the number of ZooKeeper bucket znodes but make each bucket contain more completed + event markers for the cleaner to scan. Smaller values create more bucket znodes but make + each bucket smaller. + + + + hbase.replication.bulkload.event.wait.timeout.ms + 60000 + + The maximum time, in milliseconds, that a sink RegionServer waits while another RegionServer + is processing the same replicated bulk load event. If the event is marked done before this + timeout, the retry is treated as already completed and skipped. If the timeout expires while + the in-progress marker still exists, replication fails the batch so the source can retry + later. + + + + hbase.replication.bulkload.event.wait.interval.ms + 1000 + + The sleep interval, in milliseconds, between checks while waiting for another sink + RegionServer to finish the same replicated bulk load event. Smaller values detect completion + sooner but issue more ZooKeeper checks. Larger values reduce ZooKeeper polling at the cost + of slower retry progress. + + + + hbase.master.cleaner.replication.bulkload.event.period.ms + 600000 + + The period, in milliseconds, at which the HBase Master runs the replicated bulk load event + marker cleaner. The cleaner removes expired completed-event markers after they are older + than hbase.replication.bulkload.event.done.ttl.ms and are no longer protected by a matching + in-progress marker. + + + + hbase.replication.bulkload.event.done.ttl.ms + 86400000 + + The minimum age, in milliseconds, before a completed replicated bulk load event marker can be + removed by the Master cleaner. This value should be long enough to cover expected replication + retry delays; if it is too small, a late retry may not find the completed marker and can + execute the same bulk load again. Larger values retain more ZooKeeper marker znodes. + + hbase.http.staticuser.user diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBulkLoadEventCleaner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBulkLoadEventCleaner.java index d0491c01fadd..058b5a12eed5 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBulkLoadEventCleaner.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBulkLoadEventCleaner.java @@ -17,14 +17,15 @@ */ package org.apache.hadoop.hbase.master.cleaner; +import java.io.IOException; import java.util.concurrent.TimeUnit; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.ScheduledChore; import org.apache.hadoop.hbase.Stoppable; import org.apache.hadoop.hbase.replication.regionserver.ReplicationBulkLoadEventTracker; +import org.apache.hadoop.hbase.replication.regionserver.ZKReplicationBulkLoadEventTracker; import org.apache.hadoop.hbase.zookeeper.ZKWatcher; import org.apache.yetus.audience.InterfaceAudience; -import org.apache.zookeeper.KeeperException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -48,7 +49,7 @@ public class ReplicationBulkLoadEventCleaner extends ScheduledChore { public ReplicationBulkLoadEventCleaner(Configuration conf, Stoppable stopper, ZKWatcher zkw) { super("ReplicationBulkLoadEventCleaner", stopper, conf.getInt(PERIOD_MS_KEY, PERIOD_MS_DEFAULT)); - this.tracker = new ReplicationBulkLoadEventTracker(conf, zkw); + this.tracker = new ZKReplicationBulkLoadEventTracker(conf, zkw); this.doneTtlMs = conf.getLong(DONE_TTL_MS_KEY, DONE_TTL_MS_DEFAULT); } @@ -59,12 +60,8 @@ protected void chore() { if (deleted > 0) { LOG.info("Cleaned {} replicated bulkload event marker(s)", deleted); } - } catch (KeeperException e) { + } catch (IOException e) { LOG.warn("Failed to clean replicated bulkload event markers", e); } } - - public void choreForTesting() { - chore(); - } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java index 1cfa56bb6d8f..82fa70dc73fe 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HFileReplicator.java @@ -91,7 +91,17 @@ public class HFileReplicator implements Closeable { private static final String UNDERSCORE = "_"; private final static FsPermission PERM_ALL_ACCESS = FsPermission.valueOf("-rwxrwxrwx"); private static final int COPY_BUFFER_SIZE = 65536; - // null means no throttling + private static final BulkLoadTableLoadListener NO_OP_TABLE_LOAD_LISTENER = + new BulkLoadTableLoadListener() { + @Override + public void tableLoaded(String tableName) { + } + + @Override + public void tableLoadFailed(String tableName) { + } + }; + // Double.MAX_VALUE means no throttling. private volatile RateLimiter rateLimiter; private Configuration sourceClusterConf; @@ -152,7 +162,17 @@ public void close() throws IOException { } } + interface BulkLoadTableLoadListener { + void tableLoaded(String tableName) throws IOException; + + void tableLoadFailed(String tableName); + } + public Void replicate() throws IOException { + return replicate(NO_OP_TABLE_LOAD_LISTENER); + } + + Void replicate(BulkLoadTableLoadListener tableLoadListener) throws IOException { // Copy all the hfiles to the local file system Map tableStagingDirsMap = copyHFilesToStagingDir(); @@ -174,10 +194,16 @@ public Void replicate() throws IOException { } fsDelegationToken.acquireDelegationToken(sinkFs); try { - doBulkLoad(conf, tableName, stagingDir, queue, maxRetries); + try { + doBulkLoad(conf, tableName, stagingDir, queue, maxRetries); + } catch (IOException e) { + tableLoadListener.tableLoadFailed(tableNameString); + throw e; + } } finally { cleanup(stagingDir); } + tableLoadListener.tableLoaded(tableNameString); } return null; } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationBulkLoadEventTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationBulkLoadEventTracker.java index a4d0153552dd..f3200e90bfff 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationBulkLoadEventTracker.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationBulkLoadEventTracker.java @@ -18,194 +18,33 @@ package org.apache.hadoop.hbase.replication.regionserver; import java.io.IOException; -import java.io.InterruptedIOException; import java.nio.charset.StandardCharsets; -import java.util.List; import java.util.Objects; -import java.util.concurrent.TimeUnit; -import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.TableName; -import org.apache.hadoop.hbase.util.Bytes; -import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; -import org.apache.hadoop.hbase.util.MD5Hash; -import org.apache.hadoop.hbase.zookeeper.ZKUtil; -import org.apache.hadoop.hbase.zookeeper.ZKWatcher; -import org.apache.hadoop.hbase.zookeeper.ZNodePaths; import org.apache.yetus.audience.InterfaceAudience; -import org.apache.zookeeper.KeeperException; -import org.apache.zookeeper.data.Stat; /** * Coordinates replicated bulk load events across all region servers in the sink cluster. */ @InterfaceAudience.Private -public class ReplicationBulkLoadEventTracker { +public interface ReplicationBulkLoadEventTracker { - public static final String ZNODE_KEY = "hbase.replication.bulkload.event.tracker.znode"; - public static final String ZNODE_DEFAULT = "bulkload-events"; - public static final String BUCKET_WIDTH_MS_KEY = - "hbase.replication.bulkload.event.bucket.width.ms"; - public static final long BUCKET_WIDTH_MS_DEFAULT = TimeUnit.HOURS.toMillis(1); - public static final String WAIT_TIMEOUT_MS_KEY = - "hbase.replication.bulkload.event.wait.timeout.ms"; - public static final long WAIT_TIMEOUT_MS_DEFAULT = TimeUnit.MINUTES.toMillis(1); - public static final String WAIT_INTERVAL_MS_KEY = - "hbase.replication.bulkload.event.wait.interval.ms"; - public static final long WAIT_INTERVAL_MS_DEFAULT = TimeUnit.SECONDS.toMillis(1); + Event newEvent(String replicationClusterId, TableName table, byte[] encodedRegionName, + long bulkLoadSeqNum, long writeTime); - private static final byte[] EMPTY_BYTES = new byte[0]; - private static final String IN_PROGRESS = "in-progress"; - private static final String DONE = "done"; + ClaimResult claim(Event event) throws IOException; - private final ZKWatcher zkw; - private final long bucketWidthMs; - private final long waitTimeoutMs; - private final long waitIntervalMs; - private final String inProgressZNode; - private final String doneZNode; + void markDone(Event event) throws IOException; - public ReplicationBulkLoadEventTracker(Configuration conf, ZKWatcher zkw) { - this.zkw = Objects.requireNonNull(zkw, "zkw"); - this.bucketWidthMs = Math.max(1, conf.getLong(BUCKET_WIDTH_MS_KEY, BUCKET_WIDTH_MS_DEFAULT)); - this.waitTimeoutMs = Math.max(0, conf.getLong(WAIT_TIMEOUT_MS_KEY, WAIT_TIMEOUT_MS_DEFAULT)); - this.waitIntervalMs = Math.max(1, conf.getLong(WAIT_INTERVAL_MS_KEY, WAIT_INTERVAL_MS_DEFAULT)); + void release(Event event) throws IOException; - String baseZNode = ZNodePaths.joinZNode(this.zkw.getZNodePaths().replicationZNode, - conf.get(ZNODE_KEY, ZNODE_DEFAULT)); - this.inProgressZNode = ZNodePaths.joinZNode(baseZNode, IN_PROGRESS); - this.doneZNode = ZNodePaths.joinZNode(baseZNode, DONE); - } + boolean isInProgress(Event event) throws IOException; - public Event newEvent(String replicationClusterId, TableName table, byte[] encodedRegionName, - long bulkLoadSeqNum, long writeTime) { - String bucket = Long.toString(Math.max(0, writeTime) / bucketWidthMs); - String eventKey = replicationClusterId + '\n' + table.getNameWithNamespaceInclAsString() + '\n' - + Bytes.toStringBinary(encodedRegionName) + '\n' + bulkLoadSeqNum; - return new Event(bucket, MD5Hash.getMD5AsHex(Bytes.toBytes(eventKey)), eventKey); - } + boolean isDone(Event event) throws IOException; - public ClaimResult claim(Event event) throws IOException { - long deadline = EnvironmentEdgeManager.currentTime() + waitTimeoutMs; - String inProgressPath = getInProgressPath(event); - try { - while (true) { - if (isDone(event)) { - return ClaimResult.COMPLETED; - } - ZKUtil.createWithParents(zkw, ZKUtil.getParent(inProgressPath)); - if (ZKUtil.createEphemeralNodeAndWatch(zkw, inProgressPath, event.getData())) { - return ClaimResult.CLAIMED; - } - if (isDone(event)) { - return ClaimResult.COMPLETED; - } - long now = EnvironmentEdgeManager.currentTime(); - if (now >= deadline) { - throw new IOException("Timed out waiting for replicated bulkload event " + event); - } - sleep(Math.min(waitIntervalMs, deadline - now)); - } - } catch (KeeperException e) { - throw new IOException("Failed to claim replicated bulkload event " + event, e); - } - } + int cleanDoneMarkers(long ttlMs) throws IOException; - public void markDone(Event event) throws IOException { - try { - ZKUtil.createSetData(zkw, getDonePath(event), event.getData()); - } catch (KeeperException e) { - throw new IOException("Failed to mark replicated bulkload event done " + event, e); - } - } - - public void release(Event event) throws IOException { - try { - ZKUtil.deleteNodeFailSilent(zkw, getInProgressPath(event)); - } catch (KeeperException e) { - throw new IOException("Failed to release replicated bulkload event " + event, e); - } - } - - public boolean isInProgress(Event event) throws KeeperException { - return ZKUtil.checkExists(zkw, getInProgressPath(event)) != -1; - } - - public boolean isDone(Event event) throws KeeperException { - return ZKUtil.checkExists(zkw, getDonePath(event)) != -1; - } - - public int cleanDoneMarkers(long ttlMs) throws KeeperException { - long minAgeMs = Math.max(0, ttlMs); - long now = EnvironmentEdgeManager.currentTime(); - List buckets = ZKUtil.listChildrenNoWatch(zkw, doneZNode); - if (buckets == null) { - return 0; - } - int deleted = 0; - for (String bucket : buckets) { - String doneBucketPath = ZNodePaths.joinZNode(doneZNode, bucket); - List eventIds = ZKUtil.listChildrenNoWatch(zkw, doneBucketPath); - if (eventIds == null) { - continue; - } - for (String eventId : eventIds) { - String donePath = ZNodePaths.joinZNode(doneBucketPath, eventId); - Stat stat = new Stat(); - if (ZKUtil.getDataNoWatch(zkw, donePath, stat) == null) { - continue; - } - if (now - stat.getMtime() < minAgeMs) { - continue; - } - String inProgressPath = - ZNodePaths.joinZNode(ZNodePaths.joinZNode(inProgressZNode, bucket), eventId); - if (ZKUtil.checkExists(zkw, inProgressPath) != -1) { - continue; - } - ZKUtil.deleteNodeFailSilent(zkw, donePath); - deleted++; - } - deleteIfEmpty(doneBucketPath); - deleteIfEmpty(ZNodePaths.joinZNode(inProgressZNode, bucket)); - } - return deleted; - } - - private void deleteIfEmpty(String znode) throws KeeperException { - List children = ZKUtil.listChildrenNoWatch(zkw, znode); - if (children == null || !children.isEmpty()) { - return; - } - try { - ZKUtil.deleteNodeFailSilent(zkw, znode); - } catch (KeeperException.NotEmptyException e) { - // Another region server added a child after our list; keep the bucket. - } - } - - private String getInProgressPath(Event event) { - return ZNodePaths.joinZNode(ZNodePaths.joinZNode(inProgressZNode, event.getBucket()), - event.getEventId()); - } - - private String getDonePath(Event event) { - return ZNodePaths.joinZNode(ZNodePaths.joinZNode(doneZNode, event.getBucket()), - event.getEventId()); - } - - private void sleep(long sleepMs) throws InterruptedIOException { - try { - Thread.sleep(Math.max(1, sleepMs)); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - InterruptedIOException ioe = - new InterruptedIOException("Interrupted while waiting for replicated bulkload event"); - ioe.initCause(e); - throw ioe; - } - } - - public enum ClaimResult { + enum ClaimResult { CLAIMED(true), COMPLETED(false); @@ -220,26 +59,28 @@ public boolean isClaimed() { } } - public static final class Event { + final class Event { + private static final byte[] EMPTY_BYTES = new byte[0]; + private final String bucket; private final String eventId; private final String eventKey; - private Event(String bucket, String eventId, String eventKey) { + Event(String bucket, String eventId, String eventKey) { this.bucket = bucket; this.eventId = eventId; this.eventKey = eventKey; } - public String getBucket() { + String getBucket() { return bucket; } - public String getEventId() { + String getEventId() { return eventId; } - private byte[] getData() { + byte[] getData() { return eventKey == null ? EMPTY_BYTES : eventKey.getBytes(StandardCharsets.UTF_8); } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java index 6a73b714a71e..bbb35f385eb5 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java @@ -134,15 +134,19 @@ public class ReplicationSink implements ConfigurationObserver { */ public ReplicationSink(Configuration conf, RegionServerCoprocessorHost rsServerHost) throws IOException { - this(conf, rsServerHost, null); + this(conf, rsServerHost, (ZKWatcher) null); } public ReplicationSink(Configuration conf, RegionServerCoprocessorHost rsServerHost, ZKWatcher zkw) throws IOException { + this(conf, rsServerHost, createBulkLoadEventTracker(conf, zkw)); + } + + ReplicationSink(Configuration conf, RegionServerCoprocessorHost rsServerHost, + ReplicationBulkLoadEventTracker bulkLoadEventTracker) throws IOException { this.conf = HBaseConfiguration.create(conf); this.rsServerHost = rsServerHost; - this.bulkLoadEventTracker = - zkw == null ? null : new ReplicationBulkLoadEventTracker(this.conf, zkw); + this.bulkLoadEventTracker = bulkLoadEventTracker; rowSizeWarnThreshold = conf.getInt(HConstants.BATCH_ROWS_THRESHOLD_NAME, HConstants.BATCH_ROWS_THRESHOLD_DEFAULT); replicationSinkTrackerEnabled = conf.getBoolean(REPLICATION_SINK_TRACKER_ENABLED_KEY, @@ -163,19 +167,16 @@ public ReplicationSink(Configuration conf, RegionServerCoprocessorHost rsServerH updateBulkLoadCopyBandwidth(conf); } + private static ReplicationBulkLoadEventTracker createBulkLoadEventTracker(Configuration conf, + ZKWatcher zkw) { + return zkw == null ? null : new ZKReplicationBulkLoadEventTracker(conf, zkw); + } + @Override public void onConfigurationChange(Configuration newConf) { updateBulkLoadCopyBandwidth(newConf); } - double getBulkLoadCopyRateLimiterRate() { - return bulkLoadCopyRateLimiter.getRate(); - } - - Set getInProgressBulkLoads() { - return inProgressBulkLoads; - } - private void updateBulkLoadCopyBandwidth(Configuration conf) { double bandwidthMb = conf.getDouble(HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY, HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_DEFAULT); @@ -251,8 +252,8 @@ public void replicateEntries(List entries, final ExtendedCellScanner c Map, List>> rowMap = new TreeMap<>(); Map, Map>>>> bulkLoadsPerClusters = null; - Map, List> bulkLoadEventsPerClusters = - null; + Map, Map>> bulkLoadEventsPerClustersAndTables = null; Pair, List> mutationsToWalEntriesPairs = new Pair<>(new ArrayList<>(), new ArrayList<>()); for (WALEntry entry : entries) { @@ -322,10 +323,11 @@ public void replicateEntries(List entries, final ExtendedCellScanner c bulkLoadsPerClusters.computeIfAbsent(clusterIds, k -> new HashMap<>()); buildBulkLoadHFileMap(bulkLoadHFileMap, table, bld); if (bulkLoadEvent != null) { - if (bulkLoadEventsPerClusters == null) { - bulkLoadEventsPerClusters = new HashMap<>(); + if (bulkLoadEventsPerClustersAndTables == null) { + bulkLoadEventsPerClustersAndTables = new HashMap<>(); } - bulkLoadEventsPerClusters.computeIfAbsent(clusterIds, k -> new ArrayList<>()) + bulkLoadEventsPerClustersAndTables.computeIfAbsent(clusterIds, k -> new HashMap<>()) + .computeIfAbsent(table.getNameWithNamespaceInclAsString(), k -> new ArrayList<>()) .add(bulkLoadEvent); } } @@ -397,13 +399,28 @@ public void replicateEntries(List entries, final ExtendedCellScanner c if (bulkLoadHFileMap != null && !bulkLoadHFileMap.isEmpty()) { LOG.debug("Replicating {} bulk loaded data", entry.getKey().toString()); Configuration providerConf = this.provider.getConf(this.conf, replicationClusterId); - try (HFileReplicator hFileReplicator = new HFileReplicator(providerConf, - sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, bulkLoadHFileMap, conf, - getConnection(), entry.getKey(), bulkLoadCopyRateLimiter)) { - hFileReplicator.replicate(); - markBulkLoadEventsDone(bulkLoadEventsPerClusters == null + Map> bulkLoadEventsPerTable = + bulkLoadEventsPerClustersAndTables == null ? null - : bulkLoadEventsPerClusters.get(entry.getKey()), claimedBulkLoadEvents); + : bulkLoadEventsPerClustersAndTables.get(entry.getKey()); + List releasableBulkLoadEvents = + claimedBulkLoadEvents; + try (HFileReplicator hFileReplicator = + createHFileReplicator(providerConf, sourceBaseNamespaceDirPath, + sourceHFileArchiveDirPath, bulkLoadHFileMap, entry.getKey())) { + hFileReplicator.replicate(new HFileReplicator.BulkLoadTableLoadListener() { + @Override + public void tableLoaded(String tableName) throws IOException { + markBulkLoadEventsDone(eventsForTable(bulkLoadEventsPerTable, tableName), + releasableBulkLoadEvents); + } + + @Override + public void tableLoadFailed(String tableName) { + keepBulkLoadEventsInProgress(eventsForTable(bulkLoadEventsPerTable, tableName), + releasableBulkLoadEvents); + } + }); LOG.debug("Finished replicating {} bulk loaded data", entry.getKey().toString()); } } @@ -426,6 +443,19 @@ public void replicateEntries(List entries, final ExtendedCellScanner c } } + HFileReplicator createHFileReplicator(Configuration providerConf, + String sourceBaseNamespaceDirPath, String sourceHFileArchiveDirPath, + Map>>> bulkLoadHFileMap, List sourceClusterIds) + throws IOException { + return new HFileReplicator(providerConf, sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, + bulkLoadHFileMap, conf, getConnection(), sourceClusterIds, bulkLoadCopyRateLimiter); + } + + private List eventsForTable( + Map> eventsPerTable, String tableName) { + return eventsPerTable == null ? null : eventsPerTable.get(tableName); + } + private void markBulkLoadEventsDone(List events, List releasableEvents) throws IOException { if (bulkLoadEventTracker == null || events == null) { @@ -447,11 +477,17 @@ private void markBulkLoadEventsDone(List private void keepUnmarkedBulkLoadEventsInProgress( List events, List releasableEvents, int firstUnmarkedIndex) { - if (releasableEvents == null) { + keepBulkLoadEventsInProgress(events.subList(firstUnmarkedIndex, events.size()), + releasableEvents); + } + + private void keepBulkLoadEventsInProgress(List events, + List releasableEvents) { + if (events == null || releasableEvents == null) { return; } - for (int i = firstUnmarkedIndex; i < events.size(); i++) { - releasableEvents.remove(events.get(i)); + for (ReplicationBulkLoadEventTracker.Event event : events) { + releasableEvents.remove(event); } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ZKReplicationBulkLoadEventTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ZKReplicationBulkLoadEventTracker.java new file mode 100644 index 000000000000..9ebe874e3666 --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ZKReplicationBulkLoadEventTracker.java @@ -0,0 +1,259 @@ +/* + * 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 java.io.IOException; +import java.io.InterruptedIOException; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; +import org.apache.hadoop.hbase.util.MD5Hash; +import org.apache.hadoop.hbase.zookeeper.ZKUtil; +import org.apache.hadoop.hbase.zookeeper.ZKWatcher; +import org.apache.hadoop.hbase.zookeeper.ZNodePaths; +import org.apache.yetus.audience.InterfaceAudience; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.data.Stat; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * ZooKeeper-backed replicated bulk load event tracker. + */ +@InterfaceAudience.Private +public class ZKReplicationBulkLoadEventTracker implements ReplicationBulkLoadEventTracker { + + private static final Logger LOG = + LoggerFactory.getLogger(ZKReplicationBulkLoadEventTracker.class); + + public static final String ZNODE_KEY = "hbase.replication.bulkload.event.tracker.znode"; + public static final String ZNODE_DEFAULT = "bulkload-events"; + public static final String BUCKET_WIDTH_MS_KEY = + "hbase.replication.bulkload.event.bucket.width.ms"; + public static final long BUCKET_WIDTH_MS_DEFAULT = TimeUnit.HOURS.toMillis(1); + public static final String WAIT_TIMEOUT_MS_KEY = + "hbase.replication.bulkload.event.wait.timeout.ms"; + public static final long WAIT_TIMEOUT_MS_DEFAULT = TimeUnit.MINUTES.toMillis(1); + public static final String WAIT_INTERVAL_MS_KEY = + "hbase.replication.bulkload.event.wait.interval.ms"; + public static final long WAIT_INTERVAL_MS_DEFAULT = TimeUnit.SECONDS.toMillis(1); + + private static final String IN_PROGRESS = "in-progress"; + private static final String DONE = "done"; + + private final ZKWatcher zkw; + private final long bucketWidthMs; + private final long waitTimeoutMs; + private final long waitIntervalMs; + private final String inProgressZNode; + private final String doneZNode; + + public ZKReplicationBulkLoadEventTracker(Configuration conf, ZKWatcher zkw) { + this.zkw = Objects.requireNonNull(zkw, "zkw"); + this.bucketWidthMs = Math.max(1, conf.getLong(BUCKET_WIDTH_MS_KEY, BUCKET_WIDTH_MS_DEFAULT)); + this.waitTimeoutMs = Math.max(0, conf.getLong(WAIT_TIMEOUT_MS_KEY, WAIT_TIMEOUT_MS_DEFAULT)); + this.waitIntervalMs = Math.max(1, conf.getLong(WAIT_INTERVAL_MS_KEY, WAIT_INTERVAL_MS_DEFAULT)); + + String baseZNode = ZNodePaths.joinZNode(this.zkw.getZNodePaths().replicationZNode, + conf.get(ZNODE_KEY, ZNODE_DEFAULT)); + this.inProgressZNode = ZNodePaths.joinZNode(baseZNode, IN_PROGRESS); + this.doneZNode = ZNodePaths.joinZNode(baseZNode, DONE); + } + + @Override + public ReplicationBulkLoadEventTracker.Event newEvent(String replicationClusterId, + TableName table, byte[] encodedRegionName, long bulkLoadSeqNum, long writeTime) { + String bucket = Long.toString(Math.max(0, writeTime) / bucketWidthMs); + String eventKey = replicationClusterId + '\n' + table.getNameWithNamespaceInclAsString() + '\n' + + Bytes.toStringBinary(encodedRegionName) + '\n' + bulkLoadSeqNum; + return new ReplicationBulkLoadEventTracker.Event(bucket, + MD5Hash.getMD5AsHex(Bytes.toBytes(eventKey)), eventKey); + } + + @Override + public ReplicationBulkLoadEventTracker.ClaimResult + claim(ReplicationBulkLoadEventTracker.Event event) throws IOException { + long deadline = EnvironmentEdgeManager.currentTime() + waitTimeoutMs; + String inProgressPath = getInProgressPath(event); + try { + while (true) { + if (isDoneInZK(event)) { + return ReplicationBulkLoadEventTracker.ClaimResult.COMPLETED; + } + ZKUtil.createWithParents(zkw, ZKUtil.getParent(inProgressPath)); + if (ZKUtil.createEphemeralNodeAndWatch(zkw, inProgressPath, event.getData())) { + try { + if (isDoneInZK(event)) { + deleteCreatedInProgressMarker(inProgressPath, event); + return ReplicationBulkLoadEventTracker.ClaimResult.COMPLETED; + } + return ReplicationBulkLoadEventTracker.ClaimResult.CLAIMED; + } catch (KeeperException e) { + deleteCreatedInProgressMarker(inProgressPath, event); + throw new IOException("Failed to verify replicated bulkload event completion " + event, + e); + } + } + if (isDoneInZK(event)) { + return ReplicationBulkLoadEventTracker.ClaimResult.COMPLETED; + } + long now = EnvironmentEdgeManager.currentTime(); + if (now >= deadline) { + throw new IOException("Timed out waiting for replicated bulkload event " + event); + } + sleep(Math.min(waitIntervalMs, deadline - now)); + } + } catch (KeeperException e) { + throw new IOException("Failed to claim replicated bulkload event " + event, e); + } + } + + private void deleteCreatedInProgressMarker(String inProgressPath, + ReplicationBulkLoadEventTracker.Event event) { + try { + ZKUtil.deleteNodeFailSilent(zkw, inProgressPath); + } catch (KeeperException e) { + LOG.warn("Failed to delete replicated bulkload in-progress marker {}", event, e); + } + } + + @Override + public void markDone(ReplicationBulkLoadEventTracker.Event event) throws IOException { + try { + ZKUtil.createSetData(zkw, getDonePath(event), event.getData()); + } catch (KeeperException e) { + throw new IOException("Failed to mark replicated bulkload event done " + event, e); + } + } + + @Override + public void release(ReplicationBulkLoadEventTracker.Event event) throws IOException { + try { + ZKUtil.deleteNodeFailSilent(zkw, getInProgressPath(event)); + } catch (KeeperException e) { + throw new IOException("Failed to release replicated bulkload event " + event, e); + } + } + + @Override + public boolean isInProgress(ReplicationBulkLoadEventTracker.Event event) throws IOException { + try { + return isInProgressInZK(event); + } catch (KeeperException e) { + throw new IOException("Failed to check replicated bulkload event progress " + event, e); + } + } + + @Override + public boolean isDone(ReplicationBulkLoadEventTracker.Event event) throws IOException { + try { + return isDoneInZK(event); + } catch (KeeperException e) { + throw new IOException("Failed to check replicated bulkload event completion " + event, e); + } + } + + @Override + public int cleanDoneMarkers(long ttlMs) throws IOException { + try { + long minAgeMs = Math.max(0, ttlMs); + long now = EnvironmentEdgeManager.currentTime(); + List buckets = ZKUtil.listChildrenNoWatch(zkw, doneZNode); + if (buckets == null) { + return 0; + } + int deleted = 0; + for (String bucket : buckets) { + String doneBucketPath = ZNodePaths.joinZNode(doneZNode, bucket); + List eventIds = ZKUtil.listChildrenNoWatch(zkw, doneBucketPath); + if (eventIds == null) { + continue; + } + for (String eventId : eventIds) { + String donePath = ZNodePaths.joinZNode(doneBucketPath, eventId); + Stat stat = new Stat(); + if (ZKUtil.getDataNoWatch(zkw, donePath, stat) == null) { + continue; + } + if (now - stat.getMtime() < minAgeMs) { + continue; + } + String inProgressPath = + ZNodePaths.joinZNode(ZNodePaths.joinZNode(inProgressZNode, bucket), eventId); + if (ZKUtil.checkExists(zkw, inProgressPath) != -1) { + continue; + } + ZKUtil.deleteNodeFailSilent(zkw, donePath); + deleted++; + } + deleteIfEmpty(doneBucketPath); + deleteIfEmpty(ZNodePaths.joinZNode(inProgressZNode, bucket)); + } + return deleted; + } catch (KeeperException e) { + throw new IOException("Failed to clean replicated bulkload event markers", e); + } + } + + private void deleteIfEmpty(String znode) throws KeeperException { + List children = ZKUtil.listChildrenNoWatch(zkw, znode); + if (children == null || !children.isEmpty()) { + return; + } + try { + ZKUtil.deleteNodeFailSilent(zkw, znode); + } catch (KeeperException.NotEmptyException e) { + // Another region server added a child after our list; keep the bucket. + } + } + + private boolean isInProgressInZK(ReplicationBulkLoadEventTracker.Event event) + throws KeeperException { + return ZKUtil.checkExists(zkw, getInProgressPath(event)) != -1; + } + + private boolean isDoneInZK(ReplicationBulkLoadEventTracker.Event event) throws KeeperException { + return ZKUtil.checkExists(zkw, getDonePath(event)) != -1; + } + + private String getInProgressPath(ReplicationBulkLoadEventTracker.Event event) { + return ZNodePaths.joinZNode(ZNodePaths.joinZNode(inProgressZNode, event.getBucket()), + event.getEventId()); + } + + private String getDonePath(ReplicationBulkLoadEventTracker.Event event) { + return ZNodePaths.joinZNode(ZNodePaths.joinZNode(doneZNode, event.getBucket()), + event.getEventId()); + } + + private void sleep(long sleepMs) throws InterruptedIOException { + try { + Thread.sleep(Math.max(1, sleepMs)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + InterruptedIOException ioe = + new InterruptedIOException("Interrupted while waiting for replicated bulkload event"); + ioe.initCause(e); + throw ioe; + } + } +} diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBulkLoadEventCleaner.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBulkLoadEventCleaner.java index 1282d467e1c9..9c680b1e3ff2 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBulkLoadEventCleaner.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBulkLoadEventCleaner.java @@ -17,7 +17,6 @@ */ package org.apache.hadoop.hbase.master.cleaner; -import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -26,6 +25,7 @@ import org.apache.hadoop.hbase.Stoppable; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.replication.regionserver.ReplicationBulkLoadEventTracker; +import org.apache.hadoop.hbase.replication.regionserver.ZKReplicationBulkLoadEventTracker; import org.apache.hadoop.hbase.testclassification.MediumTests; import org.apache.hadoop.hbase.testclassification.ReplicationTests; import org.apache.hadoop.hbase.util.Bytes; @@ -66,19 +66,18 @@ public static void tearDownAfterClass() throws Exception { @Test public void testCleanerRemovesExpiredDoneWithoutInProgress() throws Exception { Configuration conf = UTIL.getConfiguration(); - conf.setLong(ReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); - conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 1); + conf.setLong(ZKReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); + conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 0); - ReplicationBulkLoadEventTracker tracker = new ReplicationBulkLoadEventTracker(conf, zkw); + ReplicationBulkLoadEventTracker tracker = new ZKReplicationBulkLoadEventTracker(conf, zkw); ReplicationBulkLoadEventTracker.Event event = tracker.newEvent("cluster-A", TABLE, REGION_NAME, SEQ_NUM, WRITE_TIME); tracker.markDone(event); assertTrue(tracker.isDone(event)); - Thread.sleep(5); ReplicationBulkLoadEventCleaner cleaner = new ReplicationBulkLoadEventCleaner(conf, new NeverStopped(), zkw); - cleaner.choreForTesting(); + cleaner.chore(); assertFalse(tracker.isDone(event)); } @@ -86,19 +85,18 @@ public void testCleanerRemovesExpiredDoneWithoutInProgress() throws Exception { @Test public void testCleanerKeepsDoneWhileInProgressExists() throws Exception { Configuration conf = UTIL.getConfiguration(); - conf.setLong(ReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); - conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 1); + conf.setLong(ZKReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); + conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 0); - ReplicationBulkLoadEventTracker tracker = new ReplicationBulkLoadEventTracker(conf, zkw); + ReplicationBulkLoadEventTracker tracker = new ZKReplicationBulkLoadEventTracker(conf, zkw); ReplicationBulkLoadEventTracker.Event event = tracker.newEvent("cluster-A", TABLE, REGION_NAME, SEQ_NUM + 1, WRITE_TIME); assertTrue(tracker.claim(event).isClaimed()); tracker.markDone(event); - Thread.sleep(5); ReplicationBulkLoadEventCleaner cleaner = new ReplicationBulkLoadEventCleaner(conf, new NeverStopped(), zkw); - cleaner.choreForTesting(); + cleaner.chore(); assertTrue(tracker.isDone(event)); tracker.release(event); @@ -107,10 +105,10 @@ public void testCleanerKeepsDoneWhileInProgressExists() throws Exception { @Test public void testCleanerRemovesEmptyBucketsAfterExpiredDone() throws Exception { Configuration conf = UTIL.getConfiguration(); - conf.setLong(ReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); - conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 1); + conf.setLong(ZKReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); + conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 0); - ReplicationBulkLoadEventTracker tracker = new ReplicationBulkLoadEventTracker(conf, zkw); + ReplicationBulkLoadEventTracker tracker = new ZKReplicationBulkLoadEventTracker(conf, zkw); ReplicationBulkLoadEventTracker.Event event = tracker.newEvent("cluster-A", TABLE, REGION_NAME, SEQ_NUM + 2, WRITE_TIME); assertTrue(tracker.claim(event).isClaimed()); @@ -118,22 +116,19 @@ public void testCleanerRemovesEmptyBucketsAfterExpiredDone() throws Exception { tracker.release(event); String eventsRoot = ZNodePaths.joinZNode(zkw.getZNodePaths().replicationZNode, - ReplicationBulkLoadEventTracker.ZNODE_DEFAULT); - String doneBucketPath = - ZNodePaths.joinZNode(ZNodePaths.joinZNode(eventsRoot, "done"), event.getBucket()); - String inProgressBucketPath = - ZNodePaths.joinZNode(ZNodePaths.joinZNode(eventsRoot, "in-progress"), event.getBucket()); + ZKReplicationBulkLoadEventTracker.ZNODE_DEFAULT); + String doneRootPath = ZNodePaths.joinZNode(eventsRoot, "done"); + String inProgressRootPath = ZNodePaths.joinZNode(eventsRoot, "in-progress"); - assertTrue(ZKUtil.checkExists(zkw, doneBucketPath) != -1); - assertTrue(ZKUtil.checkExists(zkw, inProgressBucketPath) != -1); + assertFalse(ZKUtil.listChildrenNoWatch(zkw, doneRootPath).isEmpty()); + assertFalse(ZKUtil.listChildrenNoWatch(zkw, inProgressRootPath).isEmpty()); - Thread.sleep(5); ReplicationBulkLoadEventCleaner cleaner = new ReplicationBulkLoadEventCleaner(conf, new NeverStopped(), zkw); - cleaner.choreForTesting(); + cleaner.chore(); - assertEquals(-1, ZKUtil.checkExists(zkw, doneBucketPath)); - assertEquals(-1, ZKUtil.checkExists(zkw, inProgressBucketPath)); + assertTrue(ZKUtil.listChildrenNoWatch(zkw, doneRootPath).isEmpty()); + assertTrue(ZKUtil.listChildrenNoWatch(zkw, inProgressRootPath).isEmpty()); } private static final class NeverStopped implements Stoppable { diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHFileReplicatorBandwidth.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHFileReplicatorBandwidth.java index a30d7e56861e..fe9e617a8edf 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHFileReplicatorBandwidth.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHFileReplicatorBandwidth.java @@ -18,22 +18,15 @@ package org.apache.hadoop.hbase.replication.regionserver; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; -import java.io.IOException; -import java.io.OutputStream; +import java.lang.reflect.Field; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.FSDataOutputStream; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.hbase.HBaseTestingUtil; import org.apache.hadoop.hbase.testclassification.ReplicationTests; import org.apache.hadoop.hbase.testclassification.SmallTests; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.TestInstance; + +import org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter; /** * Unit tests for bulkload copy bandwidth throttling in {@link HFileReplicator} and @@ -41,98 +34,37 @@ */ @Tag(ReplicationTests.TAG) @Tag(SmallTests.TAG) -@TestInstance(TestInstance.Lifecycle.PER_CLASS) public class TestHFileReplicatorBandwidth { - private final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); - - @BeforeAll - public void setUpBeforeClass() throws Exception { - TEST_UTIL.startMiniDFSCluster(1); - } - - @AfterAll - public void tearDownAfterClass() throws Exception { - TEST_UTIL.shutdownMiniDFSCluster(); - } - /** * Verify that onConfigurationChange updates the RateLimiter rate dynamically. */ @Test public void testBandwidthDynamicUpdate() throws Exception { - Configuration conf = TEST_UTIL.getConfiguration(); + Configuration conf = new Configuration(); conf.set("hbase.replication.source.fs.conf.provider", TestSourceFSConfigurationProvider.class.getCanonicalName()); ReplicationSink sink = new ReplicationSink(conf, null); // Default: unlimited - assertEquals(Double.MAX_VALUE, sink.getBulkLoadCopyRateLimiterRate(), 0.0); + assertEquals(Double.MAX_VALUE, readBulkLoadCopyRateLimiterRate(sink), 0.0); // Apply 50 MB/s Configuration newConf = new Configuration(conf); newConf.setDouble(HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY, 50.0); sink.onConfigurationChange(newConf); - assertEquals(50.0 * 1024 * 1024, sink.getBulkLoadCopyRateLimiterRate(), 1.0); + assertEquals(50.0 * 1024 * 1024, readBulkLoadCopyRateLimiterRate(sink), 1.0); // Reset to unlimited (0 = no limit) Configuration resetConf = new Configuration(conf); resetConf.setDouble(HFileReplicator.REPLICATION_BULKLOAD_COPY_BANDWIDTH_MB_KEY, 0); sink.onConfigurationChange(resetConf); - assertEquals(Double.MAX_VALUE, sink.getBulkLoadCopyRateLimiterRate(), 0.0); - } - - /** - * Verify that throttled copy actually slows down I/O at the specified rate. Creates a file of - * known size, copies it through a throttled stream at 1 MB/s, and asserts the elapsed time is at - * least (fileSize / 1MB * 0.8) seconds. - */ - @Test - public void testCopyIsThrottled() throws Exception { - Configuration conf = TEST_UTIL.getConfiguration(); - FileSystem fs = TEST_UTIL.getTestFileSystem(); - Path testDir = TEST_UTIL.getDataTestDirOnTestFS("testCopyIsThrottled"); - fs.mkdirs(testDir); - - // Write a ~2MB test file - Path srcFile = new Path(testDir, "src"); - int fileSize = 2 * 1024 * 1024; - try (FSDataOutputStream out = fs.create(srcFile)) { - writeBytes(out, fileSize); - } - - Path dstFile = new Path(testDir, "dst"); - - // Throttle at 1 MB/s - double limitMbPerSec = 1.0; - org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter limiter = - org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter - .create(limitMbPerSec * 1024 * 1024); - - long start = System.currentTimeMillis(); - try (java.io.InputStream in = fs.open(srcFile); OutputStream out = fs.create(dstFile)) { - byte[] buf = new byte[65536]; - int bytesRead; - while ((bytesRead = in.read(buf)) >= 0) { - limiter.acquire(bytesRead); - out.write(buf, 0, bytesRead); - } - } - long elapsed = System.currentTimeMillis() - start; - - // At 1MB/s, 2MB should take >= 1600ms (allow 20% margin for JVM overhead) - long minExpectedMs = (long) (fileSize * 1000L / (limitMbPerSec * 1024 * 1024) * 0.8); - assertTrue(elapsed >= minExpectedMs, - "Expected throttled copy >= " + minExpectedMs + "ms, got " + elapsed + "ms"); + assertEquals(Double.MAX_VALUE, readBulkLoadCopyRateLimiterRate(sink), 0.0); } - private static void writeBytes(OutputStream out, int size) throws IOException { - byte[] buf = new byte[65536]; - int written = 0; - while (written < size) { - int chunk = Math.min(buf.length, size - written); - out.write(buf, 0, chunk); - written += chunk; - } + private double readBulkLoadCopyRateLimiterRate(ReplicationSink sink) throws Exception { + Field field = ReplicationSink.class.getDeclaredField("bulkLoadCopyRateLimiter"); + field.setAccessible(true); + return ((RateLimiter) field.get(sink)).getRate(); } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationBulkLoadEventTracker.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationBulkLoadEventTracker.java index 11d7dde34680..fd42715a782a 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationBulkLoadEventTracker.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationBulkLoadEventTracker.java @@ -67,11 +67,11 @@ public static void tearDownAfterClass() throws Exception { @Test public void testSecondSinkWaitsForDoneFromFirstSink() throws Exception { Configuration conf = UTIL.getConfiguration(); - conf.setLong(ReplicationBulkLoadEventTracker.WAIT_TIMEOUT_MS_KEY, 5000); - conf.setLong(ReplicationBulkLoadEventTracker.WAIT_INTERVAL_MS_KEY, 50); + conf.setLong(ZKReplicationBulkLoadEventTracker.WAIT_TIMEOUT_MS_KEY, 5000); + conf.setLong(ZKReplicationBulkLoadEventTracker.WAIT_INTERVAL_MS_KEY, 50); - ReplicationBulkLoadEventTracker tracker1 = new ReplicationBulkLoadEventTracker(conf, zkw1); - ReplicationBulkLoadEventTracker tracker2 = new ReplicationBulkLoadEventTracker(conf, zkw2); + ReplicationBulkLoadEventTracker tracker1 = new ZKReplicationBulkLoadEventTracker(conf, zkw1); + ReplicationBulkLoadEventTracker tracker2 = new ZKReplicationBulkLoadEventTracker(conf, zkw2); ReplicationBulkLoadEventTracker.Event event = tracker1.newEvent("cluster-A", TABLE, REGION_NAME, SEQ_NUM, WRITE_TIME); @@ -99,8 +99,8 @@ public void testSecondSinkWaitsForDoneFromFirstSink() throws Exception { @Test public void testDoneMarkerUsesStableBucketFromWalWriteTime() throws Exception { Configuration conf = UTIL.getConfiguration(); - conf.setLong(ReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); - ReplicationBulkLoadEventTracker tracker = new ReplicationBulkLoadEventTracker(conf, zkw1); + conf.setLong(ZKReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); + ReplicationBulkLoadEventTracker tracker = new ZKReplicationBulkLoadEventTracker(conf, zkw1); ReplicationBulkLoadEventTracker.Event first = tracker.newEvent("cluster-A", TABLE, REGION_NAME, SEQ_NUM, WRITE_TIME); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java index 13cf39406006..ab8be2b34fae 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java @@ -19,11 +19,20 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import java.io.IOException; +import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.ExtendedCell; import org.apache.hadoop.hbase.HBaseTestingUtil; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.PrivateCellUtil; @@ -31,6 +40,7 @@ import org.apache.hadoop.hbase.testclassification.MediumTests; import org.apache.hadoop.hbase.testclassification.ReplicationTests; import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.Pair; import org.apache.hadoop.hbase.wal.WALEdit; import org.apache.hadoop.hbase.wal.WALEditInternalHelper; import org.apache.hadoop.hbase.zookeeper.ZKWatcher; @@ -58,7 +68,6 @@ public class TestReplicationSinkBulkLoadDedup { private static final String CLUSTER_ID_A = "cluster-A"; - private static final String CLUSTER_ID_B = "cluster-B"; private static final byte[] REGION_NAME = Bytes.toBytes("regionXYZ"); private static final long SEQ_NUM = 100L; private static final TableName TABLE = TableName.valueOf("testDedup"); @@ -75,7 +84,7 @@ public void setUpBeforeClass() throws Exception { TestSourceFSConfigurationProvider.class.getCanonicalName()); TEST_UTIL.startMiniZKCluster(); zkw = new ZKWatcher(conf, "replication-sink-bulkload-dedup", null); - sink = new ReplicationSink(conf, null); + sink = new ReplicationSink(conf, null, zkw); } @AfterAll @@ -86,98 +95,43 @@ public void tearDownAfterClass() throws Exception { TEST_UTIL.shutdownMiniZKCluster(); } - /** - * Verify that manually adding a key to inProgressBulkLoads blocks a second add for the same key, - * and that remove restores it — core Set semantics that the dedup logic relies on. - */ - @Test - public void testInProgressSetAddAndRemove() { - String key = CLUSTER_ID_A + "#" + Bytes.toString(REGION_NAME) + "#" + SEQ_NUM; - - assertTrue(sink.getInProgressBulkLoads().add(key), "First add should succeed"); - assertFalse(sink.getInProgressBulkLoads().add(key), - "Second add should fail (already in progress)"); - - sink.getInProgressBulkLoads().remove(key); - assertTrue(sink.getInProgressBulkLoads().add(key), - "Add after remove should succeed (retry allowed)"); - sink.getInProgressBulkLoads().remove(key); // cleanup - } - - /** - * Verify that keys from different source clusters with the same region/seqNum are treated as - * distinct entries and do not block each other. - */ - @Test - public void testDifferentClustersDontConflict() { - String keyA = CLUSTER_ID_A + "#" + Bytes.toString(REGION_NAME) + "#" + SEQ_NUM; - String keyB = CLUSTER_ID_B + "#" + Bytes.toString(REGION_NAME) + "#" + SEQ_NUM; - - assertTrue(sink.getInProgressBulkLoads().add(keyA)); - assertTrue(sink.getInProgressBulkLoads().add(keyB), - "Same region/seqNum from different cluster should not conflict"); - - sink.getInProgressBulkLoads().remove(keyA); - sink.getInProgressBulkLoads().remove(keyB); - assertEquals(0, sink.getInProgressBulkLoads().size()); - } - /** * End-to-end: replicateEntries() with a bulkload WAL cell that has replicate=false should not - * register any key in inProgressBulkLoads. + * create any replicated bulk load event marker. */ @Test public void testNonReplicateBulkLoadNotTracked() throws Exception { WALProtos.BulkLoadDescriptor bld = buildBulkLoadDescriptor(REGION_NAME, SEQ_NUM, false); WALEdit edit = buildWALEdit(bld); - List entries = buildWALEntries(edit); - - int before = sink.getInProgressBulkLoads().size(); - sink.replicateEntries(entries, - PrivateCellUtil - .createExtendedCellScanner(WALEditInternalHelper.getExtendedCells(edit).iterator()), - CLUSTER_ID_A, "/dummy/namespace", "/dummy/archive"); - - assertEquals(before, sink.getInProgressBulkLoads().size(), - "Non-replicate bulkload should not add key to inProgressBulkLoads"); - } - - /** - * Simulate a concurrent retry: manually pre-populate the key to mimic a first call still in - * progress, then verify replicateEntries() skips the bulkload cell without throwing. - */ - @Test - public void testConcurrentRetryIsSkipped() throws Exception { - WALProtos.BulkLoadDescriptor bld = buildBulkLoadDescriptor(REGION_NAME, SEQ_NUM + 1, true); - String key = CLUSTER_ID_A + "#" + Bytes.toString(REGION_NAME) + "#" + (SEQ_NUM + 1); - - // Simulate first call still in progress - sink.getInProgressBulkLoads().add(key); - - WALEdit edit = buildWALEdit(bld); - List entries = buildWALEntries(edit); + long writeTime = System.currentTimeMillis(); + List entries = buildWALEntries(TABLE, REGION_NAME, edit, writeTime); + ReplicationBulkLoadEventTracker tracker = + new ZKReplicationBulkLoadEventTracker(TEST_UTIL.getConfiguration(), zkw); + ReplicationBulkLoadEventTracker.Event event = + tracker.newEvent(CLUSTER_ID_A, TABLE, REGION_NAME, SEQ_NUM, writeTime); - // Second call (retry) should skip without exception + assertFalse(tracker.isInProgress(event)); + assertFalse(tracker.isDone(event)); sink.replicateEntries(entries, PrivateCellUtil .createExtendedCellScanner(WALEditInternalHelper.getExtendedCells(edit).iterator()), CLUSTER_ID_A, "/dummy/namespace", "/dummy/archive"); - // Key still held by the "first call" - assertTrue(sink.getInProgressBulkLoads().contains(key)); - sink.getInProgressBulkLoads().remove(key); // cleanup + assertFalse(tracker.isInProgress(event)); + assertFalse(tracker.isDone(event)); } @Test public void testCompletedZkBulkLoadEventIsSkippedBySink() throws Exception { WALProtos.BulkLoadDescriptor bld = buildBulkLoadDescriptor(REGION_NAME, SEQ_NUM + 2, true); WALEdit edit = buildWALEdit(bld); - List entries = buildWALEntries(edit); + long writeTime = System.currentTimeMillis(); + List entries = buildWALEntries(TABLE, REGION_NAME, edit, writeTime); ReplicationBulkLoadEventTracker tracker = - new ReplicationBulkLoadEventTracker(TEST_UTIL.getConfiguration(), zkw); - ReplicationBulkLoadEventTracker.Event event = tracker.newEvent(CLUSTER_ID_A, TABLE, REGION_NAME, - SEQ_NUM + 2, entries.get(0).getKey().getWriteTime()); + new ZKReplicationBulkLoadEventTracker(TEST_UTIL.getConfiguration(), zkw); + ReplicationBulkLoadEventTracker.Event event = + tracker.newEvent(CLUSTER_ID_A, TABLE, REGION_NAME, SEQ_NUM + 2, writeTime); tracker.markDone(event); ReplicationSink zkSink = new ReplicationSink(TEST_UTIL.getConfiguration(), null, zkw); @@ -190,29 +144,90 @@ public void testCompletedZkBulkLoadEventIsSkippedBySink() throws Exception { assertFalse(tracker.isInProgress(event)); } + @Test + public void testLoadedEventIsMarkedDoneButFailedEventStaysInProgress() throws Exception { + TableName loadedTable = TableName.valueOf("testLoadedEvent"); + TableName failedTable = TableName.valueOf("testFailedEvent"); + long loadedSeqNum = SEQ_NUM + 3; + long failedSeqNum = SEQ_NUM + 4; + long writeTime = System.currentTimeMillis(); + RecordingBulkLoadEventTracker tracker = new RecordingBulkLoadEventTracker(); + HFileReplicator hFileReplicator = mock(HFileReplicator.class); + doAnswer(invocation -> { + HFileReplicator.BulkLoadTableLoadListener listener = invocation.getArgument(0); + listener.tableLoaded(loadedTable.getNameWithNamespaceInclAsString()); + listener.tableLoadFailed(failedTable.getNameWithNamespaceInclAsString()); + throw new IOException("failed after loading started"); + }).when(hFileReplicator).replicate(any(HFileReplicator.BulkLoadTableLoadListener.class)); + ReplicationSink testSink = + new TestableReplicationSink(TEST_UTIL.getConfiguration(), tracker, hFileReplicator); + + WALProtos.BulkLoadDescriptor loadedBld = + buildBulkLoadDescriptor(loadedTable, REGION_NAME, loadedSeqNum, true); + WALProtos.BulkLoadDescriptor failedBld = + buildBulkLoadDescriptor(failedTable, REGION_NAME, failedSeqNum, true); + WALEdit loadedEdit = buildWALEdit(loadedTable, loadedBld); + WALEdit failedEdit = buildWALEdit(failedTable, failedBld); + List entries = new ArrayList<>(); + entries.add(buildWALEntry(loadedTable, REGION_NAME, loadedEdit, writeTime)); + entries.add(buildWALEntry(failedTable, REGION_NAME, failedEdit, writeTime)); + List cells = new ArrayList<>(); + cells.addAll(WALEditInternalHelper.getExtendedCells(loadedEdit)); + cells.addAll(WALEditInternalHelper.getExtendedCells(failedEdit)); + + IOException error = assertThrows(IOException.class, + () -> testSink.replicateEntries(entries, + PrivateCellUtil.createExtendedCellScanner(cells.iterator()), CLUSTER_ID_A, + "/dummy/namespace", "/dummy/archive")); + + assertEquals("failed after loading started", error.getMessage()); + ReplicationBulkLoadEventTracker.Event loadedEvent = + tracker.getEvent(loadedTable, REGION_NAME, loadedSeqNum); + ReplicationBulkLoadEventTracker.Event failedEvent = + tracker.getEvent(failedTable, REGION_NAME, failedSeqNum); + assertTrue(tracker.doneEvents.contains(loadedEvent)); + assertTrue(tracker.releasedEvents.contains(loadedEvent)); + assertFalse(tracker.doneEvents.contains(failedEvent)); + assertFalse(tracker.releasedEvents.contains(failedEvent)); + } + // ---- helpers ---- private WALProtos.BulkLoadDescriptor buildBulkLoadDescriptor(byte[] regionName, long seqNum, boolean replicate) { + return buildBulkLoadDescriptor(TABLE, regionName, seqNum, replicate); + } + + private WALProtos.BulkLoadDescriptor buildBulkLoadDescriptor(TableName table, byte[] regionName, + long seqNum, boolean replicate) { WALProtos.StoreDescriptor store = WALProtos.StoreDescriptor.newBuilder().setFamilyName(UnsafeByteOperations.unsafeWrap(FAMILY)) .setStoreHomeDir(Bytes.toString(FAMILY)).addStoreFile("hfile-0").setStoreFileSizeBytes(1024) .build(); return WALProtos.BulkLoadDescriptor.newBuilder() - .setTableName(ProtobufUtil.toProtoTableName(TABLE)) + .setTableName(ProtobufUtil.toProtoTableName(table)) .setEncodedRegionName(UnsafeByteOperations.unsafeWrap(regionName)).addStores(store) .setBulkloadSeqNum(seqNum).setReplicate(replicate).build(); } private WALEdit buildWALEdit(WALProtos.BulkLoadDescriptor bld) { + return buildWALEdit(TABLE, bld); + } + + private WALEdit buildWALEdit(TableName table, WALProtos.BulkLoadDescriptor bld) { // RegionInfo is only used to construct the WAL cell row key; dedup logic reads // encodedRegionName from BulkLoadDescriptor directly, so any RegionInfo works here. org.apache.hadoop.hbase.client.RegionInfo ri = - org.apache.hadoop.hbase.client.RegionInfoBuilder.newBuilder(TABLE).build(); + org.apache.hadoop.hbase.client.RegionInfoBuilder.newBuilder(table).build(); return WALEdit.createBulkLoadEvent(ri, bld); } - private List buildWALEntries(WALEdit edit) { + private List buildWALEntries(TableName table, byte[] regionName, WALEdit edit, + long writeTime) { + return Collections.singletonList(buildWALEntry(table, regionName, edit, writeTime)); + } + + private WALEntry buildWALEntry(TableName table, byte[] regionName, WALEdit edit, long writeTime) { WALEntry.Builder builder = WALEntry.newBuilder(); builder.setAssociatedCellCount(edit.getCells().size()); WALKey.Builder keyBuilder = WALKey.newBuilder(); @@ -220,11 +235,86 @@ private List buildWALEntries(WALEdit edit) { uuidBuilder.setLeastSigBits(HConstants.DEFAULT_CLUSTER_ID.getLeastSignificantBits()); uuidBuilder.setMostSigBits(HConstants.DEFAULT_CLUSTER_ID.getMostSignificantBits()); keyBuilder.setClusterId(uuidBuilder.build()); - keyBuilder.setTableName(UnsafeByteOperations.unsafeWrap(TABLE.getName())); - keyBuilder.setWriteTime(System.currentTimeMillis()); - keyBuilder.setEncodedRegionName(UnsafeByteOperations.unsafeWrap(REGION_NAME)); + keyBuilder.setTableName(UnsafeByteOperations.unsafeWrap(table.getName())); + keyBuilder.setWriteTime(writeTime); + keyBuilder.setEncodedRegionName(UnsafeByteOperations.unsafeWrap(regionName)); keyBuilder.setLogSequenceNumber(-1); builder.setKey(keyBuilder.build()); - return Collections.singletonList(builder.build()); + return builder.build(); + } + + private static final class TestableReplicationSink extends ReplicationSink { + private final HFileReplicator hFileReplicator; + + TestableReplicationSink(Configuration conf, ReplicationBulkLoadEventTracker tracker, + HFileReplicator hFileReplicator) throws IOException { + super(conf, null, tracker); + this.hFileReplicator = hFileReplicator; + } + + @Override + HFileReplicator createHFileReplicator(Configuration providerConf, + String sourceBaseNamespaceDirPath, String sourceHFileArchiveDirPath, + Map>>> bulkLoadHFileMap, List sourceClusterIds) + throws IOException { + return hFileReplicator; + } + } + + private static final class RecordingBulkLoadEventTracker + implements ReplicationBulkLoadEventTracker { + private final Map events = new HashMap<>(); + private final List doneEvents = new ArrayList<>(); + private final List releasedEvents = new ArrayList<>(); + + @Override + public ReplicationBulkLoadEventTracker.Event newEvent(String replicationClusterId, + TableName table, byte[] encodedRegionName, long bulkLoadSeqNum, long writeTime) { + ReplicationBulkLoadEventTracker.Event event = new ReplicationBulkLoadEventTracker.Event("0", + table.getNameWithNamespaceInclAsString() + '#' + bulkLoadSeqNum, null); + events.put(key(table, encodedRegionName, bulkLoadSeqNum), event); + return event; + } + + @Override + public ReplicationBulkLoadEventTracker.ClaimResult + claim(ReplicationBulkLoadEventTracker.Event event) { + return ReplicationBulkLoadEventTracker.ClaimResult.CLAIMED; + } + + @Override + public void markDone(ReplicationBulkLoadEventTracker.Event event) { + doneEvents.add(event); + } + + @Override + public void release(ReplicationBulkLoadEventTracker.Event event) { + releasedEvents.add(event); + } + + @Override + public boolean isInProgress(ReplicationBulkLoadEventTracker.Event event) { + return false; + } + + @Override + public boolean isDone(ReplicationBulkLoadEventTracker.Event event) { + return doneEvents.contains(event); + } + + @Override + public int cleanDoneMarkers(long ttlMs) { + return 0; + } + + ReplicationBulkLoadEventTracker.Event getEvent(TableName table, byte[] encodedRegionName, + long bulkLoadSeqNum) { + return events.get(key(table, encodedRegionName, bulkLoadSeqNum)); + } + + private String key(TableName table, byte[] encodedRegionName, long bulkLoadSeqNum) { + return table.getNameWithNamespaceInclAsString() + '#' + + Bytes.toStringBinary(encodedRegionName) + '#' + bulkLoadSeqNum; + } } }