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/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..058b5a12eed5 --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBulkLoadEventCleaner.java @@ -0,0 +1,67 @@ +/* + * 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.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.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 ZKReplicationBulkLoadEventTracker(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 (IOException e) { + LOG.warn("Failed to clean replicated bulkload event markers", e); + } + } +} 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..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 @@ -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; @@ -78,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))); } @@ -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..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 @@ -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,32 @@ 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. 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"; + 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; + 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; private String sourceBaseNamespaceDirPath; @@ -99,6 +123,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 +152,7 @@ public HFileReplicator(Configuration sourceClusterConf, String sourceBaseNamespa REPLICATION_BULKLOAD_COPY_HFILES_PERTHREAD_DEFAULT); sinkFs = FileSystem.get(conf); + this.rateLimiter = rateLimiter; } @Override @@ -129,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(); @@ -151,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; } @@ -336,16 +385,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 +404,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/ReplicationBulkLoadEventTracker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationBulkLoadEventTracker.java new file mode 100644 index 000000000000..f3200e90bfff --- /dev/null +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationBulkLoadEventTracker.java @@ -0,0 +1,109 @@ +/* + * 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.nio.charset.StandardCharsets; +import java.util.Objects; +import org.apache.hadoop.hbase.TableName; +import org.apache.yetus.audience.InterfaceAudience; + +/** + * Coordinates replicated bulk load events across all region servers in the sink cluster. + */ +@InterfaceAudience.Private +public interface ReplicationBulkLoadEventTracker { + + Event newEvent(String replicationClusterId, TableName table, byte[] encodedRegionName, + long bulkLoadSeqNum, long writeTime); + + ClaimResult claim(Event event) throws IOException; + + void markDone(Event event) throws IOException; + + void release(Event event) throws IOException; + + boolean isInProgress(Event event) throws IOException; + + boolean isDone(Event event) throws IOException; + + int cleanDoneMarkers(long ttlMs) throws IOException; + + enum ClaimResult { + CLAIMED(true), + COMPLETED(false); + + private final boolean claimed; + + ClaimResult(boolean claimed) { + this.claimed = claimed; + } + + public boolean isClaimed() { + return claimed; + } + } + + final class Event { + private static final byte[] EMPTY_BYTES = new byte[0]; + + private final String bucket; + private final String eventId; + private final String eventKey; + + Event(String bucket, String eventId, String eventKey) { + this.bucket = bucket; + this.eventId = eventId; + this.eventKey = eventKey; + } + + String getBucket() { + return bucket; + } + + String getEventId() { + return eventId; + } + + 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 508ace390565..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 @@ -35,8 +35,10 @@ 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; @@ -59,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; @@ -66,11 +69,13 @@ 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; 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 +99,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 +113,11 @@ public class ReplicationSink { private long hfilesReplicated = 0; 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(); + private final ReplicationBulkLoadEventTracker bulkLoadEventTracker; /** * Row size threshold for multi requests above which a warning is logged @@ -124,8 +134,19 @@ public class ReplicationSink { */ public ReplicationSink(Configuration conf, RegionServerCoprocessorHost rsServerHost) throws IOException { + 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 = bulkLoadEventTracker; rowSizeWarnThreshold = conf.getInt(HConstants.BATCH_ROWS_THRESHOLD_NAME, HConstants.BATCH_ROWS_THRESHOLD_DEFAULT); replicationSinkTrackerEnabled = conf.getBoolean(REPLICATION_SINK_TRACKER_ENABLED_KEY, @@ -143,6 +164,26 @@ public ReplicationSink(Configuration conf, RegionServerCoprocessorHost rsServerH throw new IllegalArgumentException( "Configured source fs configuration provider class " + className + " throws error.", e); } + 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); + } + + 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 { @@ -199,6 +240,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 { @@ -208,6 +252,8 @@ public void replicateEntries(List entries, final ExtendedCellScanner c Map, List>> rowMap = new TreeMap<>(); Map, Map>>>> bulkLoadsPerClusters = null; + Map, Map>> bulkLoadEventsPerClustersAndTables = null; Pair, List> mutationsToWalEntriesPairs = new Pair<>(new ArrayList<>(), new ArrayList<>()); for (WALEntry entry : entries) { @@ -240,14 +286,50 @@ public void replicateEntries(List entries, final ExtendedCellScanner c if (CellUtil.matchingQualifier(cell, WALEdit.BULK_LOAD)) { BulkLoadDescriptor bld = WALEdit.getBulkLoadDescriptor(cell); if (bld.getReplicate()) { + 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 (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 = bld.getClusterIdsList(); Map>>> bulkLoadHFileMap = - bulkLoadsPerClusters.computeIfAbsent(bld.getClusterIdsList(), k -> new HashMap<>()); + bulkLoadsPerClusters.computeIfAbsent(clusterIds, k -> new HashMap<>()); buildBulkLoadHFileMap(bulkLoadHFileMap, table, bld); + if (bulkLoadEvent != null) { + if (bulkLoadEventsPerClustersAndTables == null) { + bulkLoadEventsPerClustersAndTables = new HashMap<>(); + } + bulkLoadEventsPerClustersAndTables.computeIfAbsent(clusterIds, k -> new HashMap<>()) + .computeIfAbsent(table.getNameWithNamespaceInclAsString(), k -> new ArrayList<>()) + .add(bulkLoadEvent); + } } } else if (CellUtil.matchingQualifier(cell, WALEdit.REPLICATION_MARKER)) { Mutation put = processReplicationMarkerEntry(cell); @@ -317,10 +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())) { - hFileReplicator.replicate(); + Map> bulkLoadEventsPerTable = + bulkLoadEventsPerClustersAndTables == null + ? null + : 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()); } } @@ -335,6 +435,72 @@ 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); + } + } + + 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) { + 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) { + keepBulkLoadEventsInProgress(events.subList(firstUnmarkedIndex, events.size()), + releasableEvents); + } + + private void keepBulkLoadEventsInProgress(List events, + List releasableEvents) { + if (events == null || releasableEvents == null) { + return; + } + for (ReplicationBulkLoadEventTracker.Event event : events) { + releasableEvents.remove(event); + } + } + + 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); + } } } @@ -422,6 +588,11 @@ private void addNewTableEntryInMap( bulkLoadHFileMap.put(tableName, newFamilyHFilePathsList); } + private 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/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 new file mode 100644 index 000000000000..9c680b1e3ff2 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBulkLoadEventCleaner.java @@ -0,0 +1,144 @@ +/* + * 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.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.replication.regionserver.ZKReplicationBulkLoadEventTracker; +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(ZKReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); + conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 0); + + 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)); + + ReplicationBulkLoadEventCleaner cleaner = + new ReplicationBulkLoadEventCleaner(conf, new NeverStopped(), zkw); + cleaner.chore(); + + assertFalse(tracker.isDone(event)); + } + + @Test + public void testCleanerKeepsDoneWhileInProgressExists() throws Exception { + Configuration conf = UTIL.getConfiguration(); + conf.setLong(ZKReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); + conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 0); + + 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); + + ReplicationBulkLoadEventCleaner cleaner = + new ReplicationBulkLoadEventCleaner(conf, new NeverStopped(), zkw); + cleaner.chore(); + + assertTrue(tracker.isDone(event)); + tracker.release(event); + } + + @Test + public void testCleanerRemovesEmptyBucketsAfterExpiredDone() throws Exception { + Configuration conf = UTIL.getConfiguration(); + conf.setLong(ZKReplicationBulkLoadEventTracker.BUCKET_WIDTH_MS_KEY, 1000); + conf.setLong(ReplicationBulkLoadEventCleaner.DONE_TTL_MS_KEY, 0); + + 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()); + tracker.markDone(event); + tracker.release(event); + + String eventsRoot = ZNodePaths.joinZNode(zkw.getZNodePaths().replicationZNode, + ZKReplicationBulkLoadEventTracker.ZNODE_DEFAULT); + String doneRootPath = ZNodePaths.joinZNode(eventsRoot, "done"); + String inProgressRootPath = ZNodePaths.joinZNode(eventsRoot, "in-progress"); + + assertFalse(ZKUtil.listChildrenNoWatch(zkw, doneRootPath).isEmpty()); + assertFalse(ZKUtil.listChildrenNoWatch(zkw, inProgressRootPath).isEmpty()); + + ReplicationBulkLoadEventCleaner cleaner = + new ReplicationBulkLoadEventCleaner(conf, new NeverStopped(), zkw); + cleaner.chore(); + + assertTrue(ZKUtil.listChildrenNoWatch(zkw, doneRootPath).isEmpty()); + assertTrue(ZKUtil.listChildrenNoWatch(zkw, inProgressRootPath).isEmpty()); + } + + 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/TestHFileReplicatorBandwidth.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHFileReplicatorBandwidth.java new file mode 100644 index 000000000000..fe9e617a8edf --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHFileReplicatorBandwidth.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.replication.regionserver; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.lang.reflect.Field; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.testclassification.ReplicationTests; +import org.apache.hadoop.hbase.testclassification.SmallTests; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import org.apache.hbase.thirdparty.com.google.common.util.concurrent.RateLimiter; + +/** + * 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) +public class TestHFileReplicatorBandwidth { + + /** + * Verify that onConfigurationChange updates the RateLimiter rate dynamically. + */ + @Test + public void testBandwidthDynamicUpdate() throws Exception { + 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, 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, 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, readBulkLoadCopyRateLimiterRate(sink), 0.0); + } + + 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 new file mode 100644 index 000000000000..fd42715a782a --- /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(ZKReplicationBulkLoadEventTracker.WAIT_TIMEOUT_MS_KEY, 5000); + conf.setLong(ZKReplicationBulkLoadEventTracker.WAIT_INTERVAL_MS_KEY, 50); + + 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); + + 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(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); + 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 new file mode 100644 index 000000000000..ab8be2b34fae --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestReplicationSinkBulkLoadDedup.java @@ -0,0 +1,320 @@ +/* + * 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.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; +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.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.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(MediumTests.TAG) +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class TestReplicationSinkBulkLoadDedup { + + private static final String CLUSTER_ID_A = "cluster-A"; + 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; + 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, zkw); + } + + @AfterAll + public void tearDownAfterClass() throws Exception { + if (zkw != null) { + zkw.close(); + } + TEST_UTIL.shutdownMiniZKCluster(); + } + + /** + * End-to-end: replicateEntries() with a bulkload WAL cell that has replicate=false should not + * 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); + 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); + + assertFalse(tracker.isInProgress(event)); + assertFalse(tracker.isDone(event)); + sink.replicateEntries(entries, + PrivateCellUtil + .createExtendedCellScanner(WALEditInternalHelper.getExtendedCells(edit).iterator()), + CLUSTER_ID_A, "/dummy/namespace", "/dummy/archive"); + + 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); + 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 + 2, writeTime); + 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)); + } + + @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)) + .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(); + return WALEdit.createBulkLoadEvent(ri, bld); + } + + 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(); + 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(writeTime); + keyBuilder.setEncodedRegionName(UnsafeByteOperations.unsafeWrap(regionName)); + keyBuilder.setLogSequenceNumber(-1); + builder.setKey(keyBuilder.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; + } + } +}