This is an automated email from the ASF dual-hosted git repository. wchevreuil pushed a commit to branch branch-2.1 in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-2.1 by this push: new 22ed8fb Revert "HBASE-22380 break circle replication when doing bulkload" 22ed8fb is described below commit 22ed8fbcacc34732a4a4a08b5216b192c5747210 Author: Wellington Chevreuil <wchevre...@apache.org> AuthorDate: Mon Sep 23 18:06:41 2019 +0100 Revert "HBASE-22380 break circle replication when doing bulkload" This reverts commit 11f649eefd31ba5d875a0a918313c90663a81129. --- .../apache/hadoop/hbase/regionserver/HRegion.java | 9 +- .../hadoop/hbase/regionserver/RSRpcServices.java | 10 +- .../hbase/regionserver/SecureBulkLoadManager.java | 10 +- .../replication/regionserver/HFileReplicator.java | 5 +- .../replication/regionserver/ReplicationSink.java | 44 +-- .../hadoop/hbase/tool/LoadIncrementalHFiles.java | 8 +- .../regionserver/TestBulkLoadReplication.java | 330 --------------------- 7 files changed, 23 insertions(+), 393 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java index e95e13e..f3b5a90 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java @@ -6046,7 +6046,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi */ public Map<byte[], List<Path>> bulkLoadHFiles(Collection<Pair<byte[], String>> familyPaths, boolean assignSeqId, BulkLoadListener bulkLoadListener) throws IOException { - return bulkLoadHFiles(familyPaths, assignSeqId, bulkLoadListener, false, null); + return bulkLoadHFiles(familyPaths, assignSeqId, bulkLoadListener, false); } /** @@ -6091,13 +6091,11 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi * @param bulkLoadListener Internal hooks enabling massaging/preparation of a * file about to be bulk loaded * @param copyFile always copy hfiles if true - * @param clusterIds ids from clusters that had already handled the given bulkload event. * @return Map from family to List of store file paths if successful, null if failed recoverably * @throws IOException if failed unrecoverably. */ public Map<byte[], List<Path>> bulkLoadHFiles(Collection<Pair<byte[], String>> familyPaths, - boolean assignSeqId, BulkLoadListener bulkLoadListener, - boolean copyFile, List<String> clusterIds) throws IOException { + boolean assignSeqId, BulkLoadListener bulkLoadListener, boolean copyFile) throws IOException { long seqId = -1; Map<byte[], List<Path>> storeFiles = new TreeMap<>(Bytes.BYTES_COMPARATOR); Map<String, Long> storeFilesSizes = new HashMap<>(); @@ -6272,7 +6270,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver, Regi WALProtos.BulkLoadDescriptor loadDescriptor = ProtobufUtil.toBulkLoadDescriptor(this.getRegionInfo().getTable(), UnsafeByteOperations.unsafeWrap(this.getRegionInfo().getEncodedNameAsBytes()), - storeFiles, storeFilesSizes, seqId, clusterIds); + storeFiles, + storeFilesSizes, seqId); WALUtil.writeBulkLoadMarkerAndSync(this.wal, this.getReplicationScope(), getRegionInfo(), loadDescriptor, mvcc); } catch (IOException ioe) { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java index aa54876..2b3bec7 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java @@ -2305,12 +2305,6 @@ public class RSRpcServices implements HBaseRPCErrorHandler, public BulkLoadHFileResponse bulkLoadHFile(final RpcController controller, final BulkLoadHFileRequest request) throws ServiceException { long start = EnvironmentEdgeManager.currentTime(); - List<String> clusterIds = new ArrayList<>(request.getClusterIdsList()); - if(clusterIds.contains(this.regionServer.clusterId)){ - return BulkLoadHFileResponse.newBuilder().setLoaded(true).build(); - } else { - clusterIds.add(this.regionServer.clusterId); - } try { checkOpen(); requestCount.increment(); @@ -2343,7 +2337,7 @@ public class RSRpcServices implements HBaseRPCErrorHandler, } try { map = region.bulkLoadHFiles(familyPaths, request.getAssignSeqNum(), null, - request.getCopyFile(), clusterIds); + request.getCopyFile()); } finally { if (region.getCoprocessorHost() != null) { region.getCoprocessorHost().postBulkLoadHFile(familyPaths, map); @@ -2351,7 +2345,7 @@ public class RSRpcServices implements HBaseRPCErrorHandler, } } else { // secure bulk load - map = regionServer.secureBulkLoadManager.secureBulkLoadHFiles(region, request, clusterIds); + map = regionServer.secureBulkLoadManager.secureBulkLoadHFiles(region, request); } BulkLoadHFileResponse.Builder builder = BulkLoadHFileResponse.newBuilder(); builder.setLoaded(map != null); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SecureBulkLoadManager.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SecureBulkLoadManager.java index f51608d..6b55744 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SecureBulkLoadManager.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SecureBulkLoadManager.java @@ -212,12 +212,7 @@ public class SecureBulkLoadManager { } public Map<byte[], List<Path>> secureBulkLoadHFiles(final HRegion region, - final BulkLoadHFileRequest request) throws IOException { - return secureBulkLoadHFiles(region, request, null); - } - - public Map<byte[], List<Path>> secureBulkLoadHFiles(final HRegion region, - final BulkLoadHFileRequest request, List<String> clusterIds) throws IOException { + final BulkLoadHFileRequest request) throws IOException { final List<Pair<byte[], String>> familyPaths = new ArrayList<>(request.getFamilyPathCount()); for(ClientProtos.BulkLoadHFileRequest.FamilyPath el : request.getFamilyPathList()) { familyPaths.add(new Pair<>(el.getFamily().toByteArray(), el.getPath())); @@ -293,8 +288,7 @@ public class SecureBulkLoadManager { //We call bulkLoadHFiles as requesting user //To enable access prior to staging return region.bulkLoadHFiles(familyPaths, true, - new SecureBulkLoadListener(fs, bulkToken, conf), request.getCopyFile(), - clusterIds); + new SecureBulkLoadListener(fs, bulkToken, conf), request.getCopyFile()); } catch (Exception e) { LOG.error("Failed to complete bulk load", e); } 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 c7fed77..ab9a236 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 @@ -87,19 +87,17 @@ public class HFileReplicator { private ThreadPoolExecutor exec; private int maxCopyThreads; private int copiesPerThread; - private List<String> sourceClusterIds; public HFileReplicator(Configuration sourceClusterConf, String sourceBaseNamespaceDirPath, String sourceHFileArchiveDirPath, Map<String, List<Pair<byte[], List<String>>>> tableQueueMap, Configuration conf, - Connection connection, List<String> sourceClusterIds) throws IOException { + Connection connection) throws IOException { this.sourceClusterConf = sourceClusterConf; this.sourceBaseNamespaceDirPath = sourceBaseNamespaceDirPath; this.sourceHFileArchiveDirPath = sourceHFileArchiveDirPath; this.bulkLoadHFileMap = tableQueueMap; this.conf = conf; this.connection = connection; - this.sourceClusterIds = sourceClusterIds; userProvider = UserProvider.instantiate(conf); fsDelegationToken = new FsDelegationToken(userProvider, "renewer"); @@ -130,7 +128,6 @@ public class HFileReplicator { LoadIncrementalHFiles loadHFiles = null; try { loadHFiles = new LoadIncrementalHFiles(conf); - loadHFiles.setClusterIds(sourceClusterIds); } catch (Exception e) { LOG.error("Failed to initialize LoadIncrementalHFiles for replicating bulk loaded" + " data.", e); 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 8079adc..fb4e0f9 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 @@ -174,7 +174,9 @@ public class ReplicationSink { // invocation of this method per table and cluster id. Map<TableName, Map<List<UUID>, List<Row>>> rowMap = new TreeMap<>(); - Map<List<String>, Map<String, List<Pair<byte[], List<String>>>>> bulkLoadsPerClusters = null; + // Map of table name Vs list of pair of family and list of hfile paths from its namespace + Map<String, List<Pair<byte[], List<String>>>> bulkLoadHFileMap = null; + for (WALEntry entry : entries) { TableName table = TableName.valueOf(entry.getKey().getTableName().toByteArray()); @@ -202,19 +204,10 @@ public class ReplicationSink { Cell cell = cells.current(); // Handle bulk load hfiles replication if (CellUtil.matchingQualifier(cell, WALEdit.BULK_LOAD)) { - BulkLoadDescriptor bld = WALEdit.getBulkLoadDescriptor(cell); - if(bulkLoadsPerClusters == null) { - bulkLoadsPerClusters = new HashMap<>(); - } - // Map of table name Vs list of pair of family and list of - // hfile paths from its namespace - Map<String, List<Pair<byte[], List<String>>>> bulkLoadHFileMap = - bulkLoadsPerClusters.get(bld.getClusterIdsList()); if (bulkLoadHFileMap == null) { bulkLoadHFileMap = new HashMap<>(); - bulkLoadsPerClusters.put(bld.getClusterIdsList(), bulkLoadHFileMap); } - buildBulkLoadHFileMap(bulkLoadHFileMap, table, bld); + buildBulkLoadHFileMap(bulkLoadHFileMap, table, cell); } else { // Handle wal replication if (isNewRowOrType(previousCell, cell)) { @@ -250,26 +243,14 @@ public class ReplicationSink { LOG.debug("Finished replicating mutations."); } - if(bulkLoadsPerClusters != null) { - for (Entry<List<String>, Map<String, List<Pair<byte[], - List<String>>>>> entry : bulkLoadsPerClusters.entrySet()) { - Map<String, List<Pair<byte[], List<String>>>> bulkLoadHFileMap = entry.getValue(); - if (bulkLoadHFileMap != null && !bulkLoadHFileMap.isEmpty()) { - if(LOG.isDebugEnabled()) { - LOG.debug("Started replicating bulk loaded data from cluster ids: {}.", - entry.getKey().toString()); - } - HFileReplicator hFileReplicator = - new HFileReplicator(this.provider.getConf(this.conf, replicationClusterId), + if (bulkLoadHFileMap != null && !bulkLoadHFileMap.isEmpty()) { + LOG.debug("Started replicating bulk loaded data."); + HFileReplicator hFileReplicator = + new HFileReplicator(this.provider.getConf(this.conf, replicationClusterId), sourceBaseNamespaceDirPath, sourceHFileArchiveDirPath, bulkLoadHFileMap, conf, - getConnection(), entry.getKey()); - hFileReplicator.replicate(); - if(LOG.isDebugEnabled()) { - LOG.debug("Finished replicating bulk loaded data from cluster id: {}", - entry.getKey().toString()); - } - } - } + getConnection()); + hFileReplicator.replicate(); + LOG.debug("Finished replicating bulk loaded data."); } int size = entries.size(); @@ -284,7 +265,8 @@ public class ReplicationSink { private void buildBulkLoadHFileMap( final Map<String, List<Pair<byte[], List<String>>>> bulkLoadHFileMap, TableName table, - BulkLoadDescriptor bld) throws IOException { + Cell cell) throws IOException { + BulkLoadDescriptor bld = WALEdit.getBulkLoadDescriptor(cell); List<StoreDescriptor> storesList = bld.getStoresList(); int storesSize = storesList.size(); for (int j = 0; j < storesSize; j++) { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/LoadIncrementalHFiles.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/LoadIncrementalHFiles.java index 950f2a5..e4d5dcb 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/LoadIncrementalHFiles.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/LoadIncrementalHFiles.java @@ -137,8 +137,6 @@ public class LoadIncrementalHFiles extends Configured implements Tool { private String bulkToken; - private List<String> clusterIds = new ArrayList<>(); - /** * Represents an HFile waiting to be loaded. An queue is used in this class in order to support * the case where a region has split during the process of the load. When this happens, the HFile @@ -541,7 +539,7 @@ public class LoadIncrementalHFiles extends Configured implements Tool { try (Table table = conn.getTable(getTableName())) { secureClient = new SecureBulkLoadClient(getConf(), table); success = secureClient.secureBulkLoadHFiles(getStub(), famPaths, regionName, - assignSeqIds, fsDelegationToken.getUserToken(), bulkToken, copyFile, clusterIds); + assignSeqIds, fsDelegationToken.getUserToken(), bulkToken, copyFile); } return success ? regionName : null; } finally { @@ -1253,10 +1251,6 @@ public class LoadIncrementalHFiles extends Configured implements Tool { this.bulkToken = stagingDir; } - public void setClusterIds(List<String> clusterIds) { - this.clusterIds = clusterIds; - } - /** * Infers region boundaries for a new table. * <p> diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoadReplication.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoadReplication.java deleted file mode 100644 index 4d69bdf..0000000 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoadReplication.java +++ /dev/null @@ -1,330 +0,0 @@ -/** - * 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.regionserver; - -import static org.apache.hadoop.hbase.HConstants.REPLICATION_CLUSTER_ID; -import static org.apache.hadoop.hbase.HConstants.REPLICATION_CONF_DIR; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; -import java.io.File; -import java.io.FileOutputStream; -import java.io.IOException; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.FSDataOutputStream; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.hbase.Cell; -import org.apache.hadoop.hbase.CellBuilder; -import org.apache.hadoop.hbase.CellBuilderFactory; -import org.apache.hadoop.hbase.CellBuilderType; -import org.apache.hadoop.hbase.HBaseClassTestRule; -import org.apache.hadoop.hbase.HBaseConfiguration; -import org.apache.hadoop.hbase.HBaseTestingUtility; -import org.apache.hadoop.hbase.HConstants; -import org.apache.hadoop.hbase.KeyValue; -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.Result; -import org.apache.hadoop.hbase.client.Table; -import org.apache.hadoop.hbase.client.TableDescriptor; -import org.apache.hadoop.hbase.client.TableDescriptorBuilder; -import org.apache.hadoop.hbase.coprocessor.ObserverContext; -import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor; -import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; -import org.apache.hadoop.hbase.coprocessor.RegionObserver; -import org.apache.hadoop.hbase.io.hfile.HFile; -import org.apache.hadoop.hbase.io.hfile.HFileContext; -import org.apache.hadoop.hbase.replication.ReplicationPeerConfig; -import org.apache.hadoop.hbase.replication.TestReplicationBase; -import org.apache.hadoop.hbase.testclassification.MediumTests; -import org.apache.hadoop.hbase.testclassification.ReplicationTests; -import org.apache.hadoop.hbase.tool.LoadIncrementalHFiles; -import org.apache.hadoop.hbase.util.Bytes; -import org.apache.hadoop.hbase.util.Pair; -import org.apache.hadoop.hdfs.MiniDFSCluster; -import org.junit.After; -import org.junit.Before; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Rule; -import org.junit.Test; -import org.junit.experimental.categories.Category; -import org.junit.rules.TemporaryFolder; -import org.junit.rules.TestName; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * 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 and 3). - * - * For each of defined test clusters, it performs a bulk load, asserting values on bulk loaded file - * gets replicated to other two peers. Since we are doing 3 bulk loads, with the given replication - * topology all these bulk loads should get replicated only once on each peer. To assert this, - * this test defines a preBulkLoad coprocessor and adds it to all test table regions, on each of the - * clusters. This CP counts the amount of times bulk load actually gets invoked, certifying - * we are not entering the infinite loop condition addressed by HBASE-22380. - */ -@Category({ ReplicationTests.class, MediumTests.class}) -public class TestBulkLoadReplication extends TestReplicationBase { - - @ClassRule - public static final HBaseClassTestRule CLASS_RULE = - HBaseClassTestRule.forClass(TestBulkLoadReplication.class); - - protected static final Logger LOG = - LoggerFactory.getLogger(TestBulkLoadReplication.class); - - private static final String PEER1_CLUSTER_ID = "peer1"; - private static final String PEER4_CLUSTER_ID = "peer4"; - private static final String PEER3_CLUSTER_ID = "peer3"; - - private static final String PEER_ID1 = "1"; - private static final String PEER_ID3 = "3"; - private static final String PEER_ID4 = "4"; - - private static final AtomicInteger BULK_LOADS_COUNT = new AtomicInteger(0); - private static CountDownLatch BULK_LOAD_LATCH; - - private static HBaseTestingUtility utility3; - private static HBaseTestingUtility utility4; - private static Configuration conf3; - private static Configuration conf4; - private static Table htable3; - private static Table htable4; - - @Rule - public TestName name = new TestName(); - - @ClassRule - public static TemporaryFolder testFolder = new TemporaryFolder(); - - @BeforeClass - public static void setUpBeforeClass() throws Exception { - setupBulkLoadConfigsForCluster(conf1, PEER1_CLUSTER_ID); - conf3 = HBaseConfiguration.create(conf1); - conf3.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/3"); - utility3 = new HBaseTestingUtility(conf3); - conf4 = HBaseConfiguration.create(conf1); - conf4.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/4"); - utility3 = new HBaseTestingUtility(conf3); - utility4 = new HBaseTestingUtility(conf4); - TestReplicationBase.setUpBeforeClass(); - setupBulkLoadConfigsForCluster(conf3, PEER3_CLUSTER_ID); - //utility4 is started within TestReplicationBase.setUpBeforeClass(), but we had not set - //bulkload replication configs yet, so setting a 4th utility. - setupBulkLoadConfigsForCluster(conf4, PEER4_CLUSTER_ID); - startCluster(utility3, conf3); - startCluster(utility4, conf4); - } - - private static void startCluster(HBaseTestingUtility util, Configuration configuration) - throws Exception { - LOG.info("Setup Zk to same one from utility1 and utility4"); - util.setZkCluster(utility1.getZkCluster()); - util.startMiniCluster(2); - - TableDescriptor tableDesc = TableDescriptorBuilder.newBuilder(tableName) - .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(famName).setMaxVersions(100) - .setScope(HConstants.REPLICATION_SCOPE_GLOBAL).build()) - .setColumnFamily(ColumnFamilyDescriptorBuilder.of(noRepfamName)).build(); - - Connection connection = ConnectionFactory.createConnection(configuration); - try (Admin admin = connection.getAdmin()) { - admin.createTable(tableDesc, HBaseTestingUtility.KEYS_FOR_HBA_CREATE_TABLE); - } - util.waitUntilAllRegionsAssigned(tableName); - } - - @Before - @Override - public void setUpBase() throws Exception { - super.setUpBase(); - ReplicationPeerConfig peer1Config = getPeerConfigForCluster(utility1); - ReplicationPeerConfig peer4Config = getPeerConfigForCluster(utility4); - ReplicationPeerConfig peer3Config = getPeerConfigForCluster(utility3); - //adds cluster4 as a remote peer on cluster1 - utility1.getAdmin().addReplicationPeer(PEER_ID4, peer4Config); - //adds cluster1 as a remote peer on cluster4 - utility4.getAdmin().addReplicationPeer(PEER_ID1, peer1Config); - //adds cluster3 as a remote peer on cluster4 - utility4.getAdmin().addReplicationPeer(PEER_ID3, peer3Config); - //adds cluster4 as a remote peer on cluster3 - utility3.getAdmin().addReplicationPeer(PEER_ID4, peer4Config); - setupCoprocessor(utility1); - setupCoprocessor(utility4); - setupCoprocessor(utility3); - } - - private ReplicationPeerConfig getPeerConfigForCluster(HBaseTestingUtility util) { - return ReplicationPeerConfig.newBuilder() - .setClusterKey(util.getClusterKey()).setSerial(isSerialPeer()).build(); - } - - private void setupCoprocessor(HBaseTestingUtility cluster){ - cluster.getHBaseCluster().getRegions(tableName).forEach(r -> { - try { - r.getCoprocessorHost() - .load(TestBulkLoadReplication.BulkReplicationTestObserver.class, 0, - cluster.getConfiguration()); - } catch (Exception e){ - LOG.error(e.getMessage(), e); - } - }); - } - - @After - @Override - public void tearDownBase() throws Exception { - super.tearDownBase(); - utility4.getAdmin().removeReplicationPeer(PEER_ID1); - utility4.getAdmin().removeReplicationPeer(PEER_ID3); - utility3.getAdmin().removeReplicationPeer(PEER_ID4); - } - - private static void setupBulkLoadConfigsForCluster(Configuration config, - String clusterReplicationId) throws Exception { - config.setBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY, true); - config.set(REPLICATION_CLUSTER_ID, clusterReplicationId); - File sourceConfigFolder = testFolder.newFolder(clusterReplicationId); - File sourceConfigFile = new File(sourceConfigFolder.getAbsolutePath() - + "/hbase-site.xml"); - config.writeXml(new FileOutputStream(sourceConfigFile)); - config.set(REPLICATION_CONF_DIR, testFolder.getRoot().getAbsolutePath()); - } - - @Test - public void testBulkLoadReplicationActiveActive() throws Exception { - Table peer1TestTable = utility1.getConnection().getTable(TestReplicationBase.tableName); - Table peer4TestTable = utility4.getConnection().getTable(TestReplicationBase.tableName); - Table peer3TestTable = utility3.getConnection().getTable(TestReplicationBase.tableName); - byte[] row = Bytes.toBytes("001"); - byte[] value = Bytes.toBytes("v1"); - assertBulkLoadConditions(row, value, utility1, peer1TestTable, peer4TestTable, peer3TestTable); - row = Bytes.toBytes("002"); - value = Bytes.toBytes("v2"); - assertBulkLoadConditions(row, value, utility4, peer1TestTable, peer4TestTable, peer3TestTable); - row = Bytes.toBytes("003"); - value = Bytes.toBytes("v3"); - assertBulkLoadConditions(row, value, utility3, peer1TestTable, peer4TestTable, peer3TestTable); - //Additional wait to make sure no extra bulk load happens - Thread.sleep(400); - //We have 3 bulk load events (1 initiated on each cluster). - //Each event gets 3 counts (the originator cluster, plus the two peers), - //so BULK_LOADS_COUNT expected value is 3 * 3 = 9. - assertEquals(9, BULK_LOADS_COUNT.get()); - } - - private void assertBulkLoadConditions(byte[] row, byte[] value, - HBaseTestingUtility utility, Table...tables) throws Exception { - BULK_LOAD_LATCH = new CountDownLatch(3); - bulkLoadOnCluster(row, value, utility); - assertTrue(BULK_LOAD_LATCH.await(1, TimeUnit.MINUTES)); - assertTableHasValue(tables[0], row, value); - assertTableHasValue(tables[1], row, value); - assertTableHasValue(tables[2], row, value); - } - - private void bulkLoadOnCluster(byte[] row, byte[] value, - HBaseTestingUtility cluster) throws Exception { - String bulkLoadFile = createHFileForFamilies(row, value, cluster.getConfiguration()); - Path bulkLoadFilePath = new Path(bulkLoadFile); - copyToHdfs(bulkLoadFile, cluster.getDFSCluster()); - LoadIncrementalHFiles bulkLoadHFilesTool = - new LoadIncrementalHFiles(cluster.getConfiguration()); - Map<byte[], List<Path>> family2Files = new HashMap<>(); - List<Path> files = new ArrayList<>(); - files.add(new Path("/bulk_dir/f/" + bulkLoadFilePath.getName())); - family2Files.put(Bytes.toBytes("f"), files); - bulkLoadHFilesTool.run(family2Files, tableName); - } - - private void copyToHdfs(String bulkLoadFilePath, MiniDFSCluster cluster) throws Exception { - Path bulkLoadDir = new Path("/bulk_dir/f"); - cluster.getFileSystem().mkdirs(bulkLoadDir); - cluster.getFileSystem().copyFromLocalFile(new Path(bulkLoadFilePath), bulkLoadDir); - } - - private void assertTableHasValue(Table table, byte[] row, byte[] value) throws Exception { - Get get = new Get(row); - Result result = table.get(get); - assertTrue(result.advance()); - assertEquals(Bytes.toString(value), Bytes.toString(result.value())); - } - - private String createHFileForFamilies(byte[] row, byte[] value, - Configuration clusterConfig) throws IOException { - CellBuilder cellBuilder = CellBuilderFactory.create(CellBuilderType.DEEP_COPY); - cellBuilder.setRow(row) - .setFamily(TestReplicationBase.famName) - .setQualifier(Bytes.toBytes("1")) - .setValue(value) - .setType(Cell.Type.Put); - - HFile.WriterFactory hFileFactory = HFile.getWriterFactoryNoCache(clusterConfig); - // TODO We need a way to do this without creating files - File hFileLocation = testFolder.newFile(); - FSDataOutputStream out = - new FSDataOutputStream(new FileOutputStream(hFileLocation), null); - try { - hFileFactory.withOutputStream(out); - hFileFactory.withFileContext(new HFileContext()); - HFile.Writer writer = hFileFactory.create(); - try { - writer.append(new KeyValue(cellBuilder.build())); - } finally { - writer.close(); - } - } finally { - out.close(); - } - return hFileLocation.getAbsoluteFile().getAbsolutePath(); - } - - public static class BulkReplicationTestObserver implements RegionCoprocessor { - - @Override - public Optional<RegionObserver> getRegionObserver() { - return Optional.of(new RegionObserver() { - @Override - public void preBulkLoadHFile(ObserverContext<RegionCoprocessorEnvironment> ctx, - List<Pair<byte[], String>> familyPaths) throws IOException { - BULK_LOADS_COUNT.incrementAndGet(); - } - - @Override - public void postBulkLoadHFile(ObserverContext<RegionCoprocessorEnvironment> ctx, - List<Pair<byte[], String>> stagingFamilyPaths, Map<byte[], List<Path>> finalPaths) - throws IOException { - BULK_LOAD_LATCH.countDown(); - } - }); - } - } -}