diff --git a/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java b/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java index eea169e3442..622f403f009 100644 --- a/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java +++ b/hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java @@ -1339,7 +1339,11 @@ public final class HConstants { public static final String REPLICATION_SOURCE_MAXTHREADS_KEY = "hbase.replication.source.maxthreads"; - /** Drop edits for tables that been deleted from the replication source and target */ + /** + * Drop edits for tables that been deleted from the replication source and target + * @deprecated moved it into HBaseInterClusterReplicationEndpoint + */ + @Deprecated public static final String REPLICATION_DROP_ON_DELETED_TABLE_KEY = "hbase.replication.drop.on.deleted.table"; diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALUtil.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALUtil.java index 467988226da..19f6b967358 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALUtil.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALUtil.java @@ -22,10 +22,12 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Map; import java.util.NavigableMap; +import java.util.function.Function; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.client.RegionInfo; import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl; import org.apache.hadoop.hbase.util.CommonFSUtils; @@ -195,4 +197,23 @@ public class WALUtil { } return conf.getLong(WAL_BLOCK_SIZE, defaultBlockSize); } + + public static void filterCells(WALEdit edit, Function mapper) { + ArrayList cells = edit.getCells(); + int size = cells.size(); + int newSize = 0; + for (int i = 0; i < size; i++) { + Cell cell = mapper.apply(cells.get(i)); + if (cell != null) { + cells.set(newSize, cell); + newSize++; + } + } + for (int i = size - 1; i >= newSize; i--) { + cells.remove(i); + } + if (newSize < size / 2) { + cells.trimToSize(); + } + } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java index 2731de8b032..9539d30eddf 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/HBaseInterClusterReplicationEndpoint.java @@ -24,8 +24,11 @@ import java.net.SocketTimeoutException; import java.net.UnknownHostException; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.TreeMap; import java.util.concurrent.Callable; import java.util.concurrent.CompletionService; @@ -34,40 +37,44 @@ import java.util.concurrent.ExecutorCompletionService; import java.util.concurrent.Future; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; -import java.util.regex.Matcher; -import java.util.regex.Pattern; import java.util.stream.Collectors; import java.util.stream.Stream; + import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.Abortable; +import org.apache.hadoop.hbase.CellUtil; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HConstants; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.TableNotFoundException; +import org.apache.hadoop.hbase.client.Admin; import org.apache.hadoop.hbase.client.ClusterConnection; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.ipc.RpcServer; import org.apache.hadoop.hbase.protobuf.ReplicationProtbufUtil; +import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException; +import org.apache.hadoop.hbase.regionserver.wal.WALUtil; import org.apache.hadoop.hbase.replication.HBaseReplicationEndpoint; import org.apache.hadoop.hbase.replication.regionserver.ReplicationSinkManager.SinkPeer; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.CommonFSUtils; import org.apache.hadoop.hbase.util.Threads; import org.apache.hadoop.hbase.wal.WAL.Entry; +import org.apache.hadoop.hbase.wal.WALEdit; import org.apache.hadoop.ipc.RemoteException; import org.apache.yetus.audience.InterfaceAudience; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService.BlockingInterface; + import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; import org.apache.hbase.thirdparty.com.google.common.base.Preconditions; import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder; -import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService.BlockingInterface; - /** * A {@link org.apache.hadoop.hbase.replication.ReplicationEndpoint} * implementation for replicating to another HBase cluster. @@ -86,6 +93,13 @@ public class HBaseInterClusterReplicationEndpoint extends HBaseReplicationEndpoi private static final long DEFAULT_MAX_TERMINATION_WAIT_MULTIPLIER = 2; + /** Drop edits for tables that been deleted from the replication source and target */ + public static final String REPLICATION_DROP_ON_DELETED_TABLE_KEY = + "hbase.replication.drop.on.deleted.table"; + /** Drop edits for CFs that been deleted from the replication source and target */ + public static final String REPLICATION_DROP_ON_DELETED_COLUMN_FAMILY_KEY = + "hbase.replication.drop.on.deleted.columnfamily"; + private ClusterConnection conn; private Configuration localConf; private Configuration conf; @@ -112,6 +126,7 @@ public class HBaseInterClusterReplicationEndpoint extends HBaseReplicationEndpoi private boolean replicationBulkLoadDataEnabled; private Abortable abortable; private boolean dropOnDeletedTables; + private boolean dropOnDeletedColumnFamilies; private boolean isSerial = false; /* @@ -174,7 +189,9 @@ public class HBaseInterClusterReplicationEndpoint extends HBaseReplicationEndpoi this.replicationRpcLimit = (int)(0.95 * conf.getLong(RpcServer.MAX_REQUEST_SIZE, RpcServer.DEFAULT_MAX_REQUEST_SIZE)); this.dropOnDeletedTables = - this.conf.getBoolean(HConstants.REPLICATION_DROP_ON_DELETED_TABLE_KEY, false); + this.conf.getBoolean(REPLICATION_DROP_ON_DELETED_TABLE_KEY, false); + this.dropOnDeletedColumnFamilies = this.conf + .getBoolean(REPLICATION_DROP_ON_DELETED_COLUMN_FAMILY_KEY, false); this.replicationBulkLoadDataEnabled = conf.getBoolean(HConstants.REPLICATION_BULKLOAD_ENABLE_KEY, @@ -285,28 +302,148 @@ public class HBaseInterClusterReplicationEndpoint extends HBaseReplicationEndpoi } } - private TableName parseTable(String msg) { - // ... TableNotFoundException: ''/n... - Pattern p = Pattern.compile("TableNotFoundException: '([\\S]*)'"); - Matcher m = p.matcher(msg); - if (m.find()) { - String table = m.group(1); - try { - // double check that table is a valid table name - TableName.valueOf(TableName.isLegalFullyQualifiedTableName(Bytes.toBytes(table))); - return TableName.valueOf(table); - } catch (IllegalArgumentException ignore) { + /** + * Check if there's an {@link TableNotFoundException} in the caused by stacktrace. + */ + @VisibleForTesting + public static boolean isTableNotFoundException(Throwable io) { + if (io instanceof RemoteException) { + io = ((RemoteException) io).unwrapRemoteException(); + } + if (io != null && io.getMessage().contains("TableNotFoundException")) { + return true; + } + for (; io != null; io = io.getCause()) { + if (io instanceof TableNotFoundException) { + return true; } } - return null; + return false; } - // Filter a set of batches by TableName - private List> filterBatches(final List> oldEntryList, TableName table) { - return oldEntryList - .stream().map(entries -> entries.stream() - .filter(e -> !e.getKey().getTableName().equals(table)).collect(Collectors.toList())) - .collect(Collectors.toList()); + /** + * Check if there's an {@link NoSuchColumnFamilyException} in the caused by stacktrace. + */ + @VisibleForTesting + public static boolean isNoSuchColumnFamilyException(Throwable io) { + if (io instanceof RemoteException) { + io = ((RemoteException) io).unwrapRemoteException(); + } + if (io != null && io.getMessage().contains("NoSuchColumnFamilyException")) { + return true; + } + for (; io != null; io = io.getCause()) { + if (io instanceof NoSuchColumnFamilyException) { + return true; + } + } + return false; + } + + @VisibleForTesting + List> filterNotExistTableEdits(final List> oldEntryList) { + List> entryList = new ArrayList<>(); + Map existMap = new HashMap<>(); + try (Connection localConn = ConnectionFactory.createConnection(ctx.getLocalConfiguration()); + Admin localAdmin = localConn.getAdmin()) { + for (List oldEntries : oldEntryList) { + List entries = new ArrayList<>(); + for (Entry e : oldEntries) { + TableName tableName = e.getKey().getTableName(); + boolean exist = true; + if (existMap.containsKey(tableName)) { + exist = existMap.get(tableName); + } else { + try { + exist = localAdmin.tableExists(tableName); + existMap.put(tableName, exist); + } catch (IOException iox) { + LOG.warn("Exception checking for local table " + tableName, iox); + // we can't drop edits without full assurance, so we assume table exists. + exist = true; + } + } + if (exist) { + entries.add(e); + } else { + // Would potentially be better to retry in one of the outer loops + // and add a table filter there; but that would break the encapsulation, + // so we're doing the filtering here. + LOG.warn("Missing table detected at sink, local table also does not exist, " + + "filtering edits for table '{}'", tableName); + } + } + if (!entries.isEmpty()) { + entryList.add(entries); + } + } + } catch (IOException iox) { + LOG.warn("Exception when creating connection to check local table", iox); + return oldEntryList; + } + return entryList; + } + + @VisibleForTesting + List> filterNotExistColumnFamilyEdits(final List> oldEntryList) { + List> entryList = new ArrayList<>(); + Map> existColumnFamilyMap = new HashMap<>(); + try (Connection localConn = ConnectionFactory.createConnection(ctx.getLocalConfiguration()); + Admin localAdmin = localConn.getAdmin()) { + for (List oldEntries : oldEntryList) { + List entries = new ArrayList<>(); + for (Entry e : oldEntries) { + TableName tableName = e.getKey().getTableName(); + if (!existColumnFamilyMap.containsKey(tableName)) { + try { + Set cfs = localAdmin.getDescriptor(tableName).getColumnFamilyNames().stream() + .map(Bytes::toString).collect(Collectors.toSet()); + existColumnFamilyMap.put(tableName, cfs); + } catch (Exception ex) { + LOG.warn("Exception getting cf names for local table {}", tableName, ex); + // if catch any exception, we are not sure about table's description, + // so replicate raw entry + entries.add(e); + continue; + } + } + + Set existColumnFamilies = existColumnFamilyMap.get(tableName); + Set missingCFs = new HashSet<>(); + WALEdit walEdit = new WALEdit(); + walEdit.getCells().addAll(e.getEdit().getCells()); + WALUtil.filterCells(walEdit, cell -> { + String cf = Bytes.toString(CellUtil.cloneFamily(cell)); + if (existColumnFamilies.contains(cf)) { + return cell; + } else { + missingCFs.add(cf); + return null; + } + }); + if (!walEdit.isEmpty()) { + Entry newEntry = new Entry(e.getKey(), walEdit); + entries.add(newEntry); + } + + if (!missingCFs.isEmpty()) { + // Would potentially be better to retry in one of the outer loops + // and add a table filter there; but that would break the encapsulation, + // so we're doing the filtering here. + LOG.warn( + "Missing column family detected at sink, local column family also does not exist," + + " filtering edits for table '{}',column family '{}'", tableName, missingCFs); + } + } + if (!entries.isEmpty()) { + entryList.add(entries); + } + } + } catch (IOException iox) { + LOG.warn("Exception when creating connection to check local table", iox); + return oldEntryList; + } + return entryList; } private void reconnectToPeerCluster() { @@ -403,36 +540,21 @@ public class HBaseInterClusterReplicationEndpoint extends HBaseReplicationEndpoi return true; } catch (IOException ioe) { if (ioe instanceof RemoteException) { - ioe = ((RemoteException) ioe).unwrapRemoteException(); - LOG.warn("{} Can't replicate because of an error on the remote cluster: ", logPeerId(), - ioe); - if (ioe instanceof TableNotFoundException) { - if (dropOnDeletedTables) { - // this is a bit fragile, but cannot change how TNFE is serialized - // at least check whether the table name is legal - TableName table = parseTable(ioe.getMessage()); - if (table != null) { - try (Connection localConn = - ConnectionFactory.createConnection(ctx.getLocalConfiguration())) { - if (!localConn.getAdmin().tableExists(table)) { - // Would potentially be better to retry in one of the outer loops - // and add a table filter there; but that would break the encapsulation, - // so we're doing the filtering here. - LOG.info("{} Missing table detected at sink, local table also does not " - + "exist, filtering edits for '{}'", logPeerId(), table); - batches = filterBatches(batches, table); - continue; - } - } catch (IOException iox) { - LOG.warn("{} Exception checking for local table: ", logPeerId(), iox); - } - } + if (dropOnDeletedTables && isTableNotFoundException(ioe)) { + // Only filter the edits to replicate and don't change the entries in replicateContext + // as the upper layer rely on it. + batches = filterNotExistTableEdits(batches); + if (batches.isEmpty()) { + LOG.warn("After filter not exist table's edits, 0 edits to replicate, just return"); + return true; + } + } else if (dropOnDeletedColumnFamilies && isNoSuchColumnFamilyException(ioe)) { + batches = filterNotExistColumnFamilyEdits(batches); + if (batches.isEmpty()) { + LOG.warn( + "After filter not exist column family's edits, 0 edits to replicate, just return"); + return true; } - // fall through and sleep below - } else { - LOG.warn("{} Peer encountered RemoteException, rechecking all sinks: ", logPeerId(), - ioe); - replicationSinkMgr.chooseSinks(); } } else { if (ioe instanceof SocketTimeoutException) { diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationDroppedTables.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationDroppedTables.java index 2d039b0674b..872aa0a204d 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationDroppedTables.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationDroppedTables.java @@ -17,6 +17,7 @@ */ package org.apache.hadoop.hbase.replication; +import static org.apache.hadoop.hbase.replication.regionserver.HBaseInterClusterReplicationEndpoint.REPLICATION_DROP_ON_DELETED_TABLE_KEY; import static org.junit.Assert.fail; import java.io.IOException; @@ -136,7 +137,7 @@ public class TestReplicationDroppedTables extends TestReplicationBase { } private void testEditsBehindDroppedTable(boolean allowProceeding, String tName) throws Exception { - CONF1.setBoolean(HConstants.REPLICATION_DROP_ON_DELETED_TABLE_KEY, allowProceeding); + CONF1.setBoolean(REPLICATION_DROP_ON_DELETED_TABLE_KEY, allowProceeding); CONF1.setInt(HConstants.REPLICATION_SOURCE_MAXTHREADS_KEY, 1); // make sure we have a single region server only, so that all @@ -205,12 +206,12 @@ public class TestReplicationDroppedTables extends TestReplicationBase { verifyReplicationStuck(); } // just to be safe - CONF1.setBoolean(HConstants.REPLICATION_DROP_ON_DELETED_TABLE_KEY, false); + CONF1.setBoolean(REPLICATION_DROP_ON_DELETED_TABLE_KEY, false); } @Test public void testEditsBehindDroppedTableTiming() throws Exception { - CONF1.setBoolean(HConstants.REPLICATION_DROP_ON_DELETED_TABLE_KEY, true); + CONF1.setBoolean(REPLICATION_DROP_ON_DELETED_TABLE_KEY, true); CONF1.setInt(HConstants.REPLICATION_SOURCE_MAXTHREADS_KEY, 1); // make sure we have a single region server only, so that all @@ -281,7 +282,7 @@ public class TestReplicationDroppedTables extends TestReplicationBase { verifyReplicationProceeded(); } // just to be safe - CONF1.setBoolean(HConstants.REPLICATION_DROP_ON_DELETED_TABLE_KEY, false); + CONF1.setBoolean(REPLICATION_DROP_ON_DELETED_TABLE_KEY, false); } private boolean peerHasAllNormalRows() throws IOException { diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationEditsDroppedWithDeletedTableCFs.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationEditsDroppedWithDeletedTableCFs.java new file mode 100644 index 00000000000..55ef82552c9 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationEditsDroppedWithDeletedTableCFs.java @@ -0,0 +1,240 @@ +/** + * 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; + +import static org.apache.hadoop.hbase.HConstants.REPLICATION_SCOPE_GLOBAL; +import static org.apache.hadoop.hbase.HConstants.ZOOKEEPER_ZNODE_PARENT; +import static org.apache.hadoop.hbase.replication.regionserver.HBaseInterClusterReplicationEndpoint.REPLICATION_DROP_ON_DELETED_COLUMN_FAMILY_KEY; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.util.Arrays; +import java.util.stream.Collectors; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseClassTestRule; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HBaseTestingUtility; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.Waiter.Predicate; +import org.apache.hadoop.hbase.client.Admin; +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Put; +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.testclassification.LargeTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.JVMClusterUtil; +import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@Category({ LargeTests.class }) +public class TestReplicationEditsDroppedWithDeletedTableCFs { + + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestReplicationEditsDroppedWithDeletedTableCFs.class); + + private static final Logger LOG = + LoggerFactory.getLogger(TestReplicationEditsDroppedWithDeletedTableCFs.class); + + private static Configuration conf1 = HBaseConfiguration.create(); + private static Configuration conf2 = HBaseConfiguration.create(); + + protected static HBaseTestingUtility utility1; + protected static HBaseTestingUtility utility2; + + private static Admin admin1; + private static Admin admin2; + + private static final TableName TABLE = TableName.valueOf("table"); + private static final byte[] NORMAL_CF = Bytes.toBytes("normal_cf"); + private static final byte[] DROPPED_CF = Bytes.toBytes("dropped_cf"); + + private static final byte[] ROW = Bytes.toBytes("row"); + private static final byte[] QUALIFIER = Bytes.toBytes("q"); + private static final byte[] VALUE = Bytes.toBytes("value"); + + private static final String PEER_ID = "1"; + private static final long SLEEP_TIME = 1000; + private static final int NB_RETRIES = 10; + + @BeforeClass + public static void setUpBeforeClass() throws Exception { + // Set true to filter replication edits for dropped table + conf1.setBoolean(REPLICATION_DROP_ON_DELETED_COLUMN_FAMILY_KEY, true); + conf1.set(ZOOKEEPER_ZNODE_PARENT, "/1"); + conf1.setInt("replication.source.nb.capacity", 1); + utility1 = new HBaseTestingUtility(conf1); + utility1.startMiniZKCluster(); + MiniZooKeeperCluster miniZK = utility1.getZkCluster(); + conf1 = utility1.getConfiguration(); + + conf2 = HBaseConfiguration.create(conf1); + conf2.set(ZOOKEEPER_ZNODE_PARENT, "/2"); + utility2 = new HBaseTestingUtility(conf2); + utility2.setZkCluster(miniZK); + + utility1.startMiniCluster(1); + utility2.startMiniCluster(1); + + admin1 = utility1.getAdmin(); + admin2 = utility2.getAdmin(); + } + + @AfterClass + public static void tearDownAfterClass() throws Exception { + utility2.shutdownMiniCluster(); + utility1.shutdownMiniCluster(); + } + + @Before + public void setup() throws Exception { + // Roll log + for (JVMClusterUtil.RegionServerThread r : utility1.getHBaseCluster() + .getRegionServerThreads()) { + utility1.getAdmin().rollWALWriter(r.getRegionServer().getServerName()); + } + // add peer + ReplicationPeerConfig rpc = ReplicationPeerConfig.newBuilder() + .setClusterKey(utility2.getClusterKey()) + .setReplicateAllUserTables(true).build(); + admin1.addReplicationPeer(PEER_ID, rpc); + // create table + createTable(); + } + + @After + public void tearDown() throws Exception { + // Remove peer + admin1.removeReplicationPeer(PEER_ID); + // Drop table + admin1.disableTable(TABLE); + admin1.deleteTable(TABLE); + admin2.disableTable(TABLE); + admin2.deleteTable(TABLE); + } + + private void createTable() throws Exception { + TableDescriptor desc = createTableDescriptor(NORMAL_CF, DROPPED_CF); + admin1.createTable(desc); + admin2.createTable(desc); + utility1.waitUntilAllRegionsAssigned(desc.getTableName()); + utility2.waitUntilAllRegionsAssigned(desc.getTableName()); + } + + @Test + public void testEditsDroppedWithDeleteCF() throws Exception { + admin1.disableReplicationPeer(PEER_ID); + + try (Table table = utility1.getConnection().getTable(TABLE)) { + Put put = new Put(ROW); + put.addColumn(DROPPED_CF, QUALIFIER, VALUE); + table.put(put); + } + + deleteCf(admin1); + deleteCf(admin2); + + admin1.enableReplicationPeer(PEER_ID); + + verifyReplicationProceeded(); + } + + @Test + public void testEditsBehindDeleteCFTiming() throws Exception { + admin1.disableReplicationPeer(PEER_ID); + + try (Table table = utility1.getConnection().getTable(TABLE)) { + Put put = new Put(ROW); + put.addColumn(DROPPED_CF, QUALIFIER, VALUE); + table.put(put); + } + + // Only delete cf from peer cluster + deleteCf(admin2); + + admin1.enableReplicationPeer(PEER_ID); + + // the source table's cf still exists, replication should be stalled + verifyReplicationStuck(); + deleteCf(admin1); + // now the source table's cf is gone, replication should proceed, the + // offending edits be dropped + verifyReplicationProceeded(); + } + + private void verifyReplicationProceeded() throws Exception { + try (Table table = utility1.getConnection().getTable(TABLE)) { + Put put = new Put(ROW); + put.addColumn(NORMAL_CF, QUALIFIER, VALUE); + table.put(put); + } + utility2.waitFor(NB_RETRIES * SLEEP_TIME, (Predicate) () -> { + try (Table peerTable = utility2.getConnection().getTable(TABLE)) { + Result result = peerTable.get(new Get(ROW).addColumn(NORMAL_CF, QUALIFIER)); + return result != null && !result.isEmpty() + && Bytes.equals(VALUE, result.getValue(NORMAL_CF, QUALIFIER)); + } + }); + } + + private void verifyReplicationStuck() throws Exception { + try (Table table = utility1.getConnection().getTable(TABLE)) { + Put put = new Put(ROW); + put.addColumn(NORMAL_CF, QUALIFIER, VALUE); + table.put(put); + } + try (Table peerTable = utility2.getConnection().getTable(TABLE)) { + for (int i = 0; i < NB_RETRIES; i++) { + Result result = peerTable.get(new Get(ROW).addColumn(NORMAL_CF, QUALIFIER)); + if (result != null && !result.isEmpty()) { + fail("Edit should have been stuck behind dropped tables, but value is " + Bytes + .toString(result.getValue(NORMAL_CF, QUALIFIER))); + } else { + LOG.info("Row not replicated, let's wait a bit more..."); + Thread.sleep(SLEEP_TIME); + } + } + } + } + + private TableDescriptor createTableDescriptor(byte[]... cfs) { + return TableDescriptorBuilder.newBuilder(TABLE) + .setColumnFamilies(Arrays.stream(cfs).map(cf -> + ColumnFamilyDescriptorBuilder.newBuilder(cf).setScope(REPLICATION_SCOPE_GLOBAL).build()) + .collect(Collectors.toList()) + ).build(); + } + + private void deleteCf(Admin admin) throws IOException { + TableDescriptor desc = createTableDescriptor(NORMAL_CF); + admin.modifyTable(desc); + } +} \ No newline at end of file diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationEditsDroppedWithDroppedTable.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationEditsDroppedWithDroppedTable.java new file mode 100644 index 00000000000..56eecacd99c --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationEditsDroppedWithDroppedTable.java @@ -0,0 +1,249 @@ +/** + * 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; + +import static org.junit.Assert.fail; + +import org.apache.hadoop.conf.Configuration; +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.NamespaceDescriptor; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.Waiter.Predicate; +import org.apache.hadoop.hbase.client.Admin; +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Put; +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.replication.regionserver.HBaseInterClusterReplicationEndpoint; +import org.apache.hadoop.hbase.testclassification.LargeTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.util.JVMClusterUtil; +import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@Category({ LargeTests.class }) +public class TestReplicationEditsDroppedWithDroppedTable { + + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestReplicationEditsDroppedWithDroppedTable.class); + + private static final Logger LOG = + LoggerFactory.getLogger(TestReplicationEditsDroppedWithDroppedTable.class); + + private static Configuration conf1 = HBaseConfiguration.create(); + private static Configuration conf2 = HBaseConfiguration.create(); + + protected static HBaseTestingUtility utility1; + protected static HBaseTestingUtility utility2; + + private static Admin admin1; + private static Admin admin2; + + private static final String namespace = "NS"; + private static final TableName NORMAL_TABLE = TableName.valueOf("normal-table"); + private static final TableName DROPPED_TABLE = TableName.valueOf("dropped-table"); + private static final TableName DROPPED_NS_TABLE = TableName.valueOf("NS:dropped-table"); + private static final byte[] ROW = Bytes.toBytes("row"); + private static final byte[] FAMILY = Bytes.toBytes("f"); + private static final byte[] QUALIFIER = Bytes.toBytes("q"); + private static final byte[] VALUE = Bytes.toBytes("value"); + + private static final String PEER_ID = "1"; + private static final long SLEEP_TIME = 1000; + private static final int NB_RETRIES = 10; + + @BeforeClass + public static void setUpBeforeClass() throws Exception { + // Set true to filter replication edits for dropped table + conf1.setBoolean(HBaseInterClusterReplicationEndpoint.REPLICATION_DROP_ON_DELETED_TABLE_KEY, + true); + conf1.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/1"); + conf1.setInt("replication.source.nb.capacity", 1); + utility1 = new HBaseTestingUtility(conf1); + utility1.startMiniZKCluster(); + MiniZooKeeperCluster miniZK = utility1.getZkCluster(); + conf1 = utility1.getConfiguration(); + + conf2 = HBaseConfiguration.create(conf1); + conf2.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/2"); + utility2 = new HBaseTestingUtility(conf2); + utility2.setZkCluster(miniZK); + + utility1.startMiniCluster(1); + utility2.startMiniCluster(1); + + admin1 = utility1.getAdmin(); + admin2 = utility2.getAdmin(); + + NamespaceDescriptor nsDesc = NamespaceDescriptor.create(namespace).build(); + admin1.createNamespace(nsDesc); + admin2.createNamespace(nsDesc); + } + + @AfterClass + public static void tearDownAfterClass() throws Exception { + utility2.shutdownMiniCluster(); + utility1.shutdownMiniCluster(); + } + + @Before + public void setup() throws Exception { + // Roll log + for (JVMClusterUtil.RegionServerThread r : utility1.getHBaseCluster() + .getRegionServerThreads()) { + utility1.getAdmin().rollWALWriter(r.getRegionServer().getServerName()); + } + // add peer + ReplicationPeerConfig rpc = ReplicationPeerConfig.newBuilder() + .setClusterKey(utility2.getClusterKey()) + .setReplicateAllUserTables(true).build(); + admin1.addReplicationPeer(PEER_ID, rpc); + // create table + createTable(NORMAL_TABLE); + } + + @After + public void tearDown() throws Exception { + // Remove peer + admin1.removeReplicationPeer(PEER_ID); + // Drop table + admin1.disableTable(NORMAL_TABLE); + admin1.deleteTable(NORMAL_TABLE); + admin2.disableTable(NORMAL_TABLE); + admin2.deleteTable(NORMAL_TABLE); + } + + private void createTable(TableName tableName) throws Exception { + TableDescriptor desc = TableDescriptorBuilder.newBuilder(tableName).setColumnFamily( + ColumnFamilyDescriptorBuilder.newBuilder(FAMILY) + .setScope(HConstants.REPLICATION_SCOPE_GLOBAL).build() + ).build(); + admin1.createTable(desc); + admin2.createTable(desc); + utility1.waitUntilAllRegionsAssigned(tableName); + utility2.waitUntilAllRegionsAssigned(tableName); + } + + @Test + public void testEditsDroppedWithDroppedTable() throws Exception { + testWithDroppedTable(DROPPED_TABLE); + } + + @Test + public void testEditsDroppedWithDroppedTableNS() throws Exception { + testWithDroppedTable(DROPPED_NS_TABLE); + } + + private void testWithDroppedTable(TableName droppedTableName) throws Exception { + createTable(droppedTableName); + admin1.disableReplicationPeer(PEER_ID); + + try (Table droppedTable = utility1.getConnection().getTable(droppedTableName)) { + Put put = new Put(ROW); + put.addColumn(FAMILY, QUALIFIER, VALUE); + droppedTable.put(put); + } + + admin1.disableTable(droppedTableName); + admin1.deleteTable(droppedTableName); + admin2.disableTable(droppedTableName); + admin2.deleteTable(droppedTableName); + + admin1.enableReplicationPeer(PEER_ID); + + verifyReplicationProceeded(); + } + + @Test + public void testEditsBehindDroppedTableTiming() throws Exception { + createTable(DROPPED_TABLE); + admin1.disableReplicationPeer(PEER_ID); + + try (Table droppedTable = utility1.getConnection().getTable(DROPPED_TABLE)) { + Put put = new Put(ROW); + put.addColumn(FAMILY, QUALIFIER, VALUE); + droppedTable.put(put); + } + + // Only delete table from peer cluster + admin2.disableTable(DROPPED_TABLE); + admin2.deleteTable(DROPPED_TABLE); + + admin1.enableReplicationPeer(PEER_ID); + + // the source table still exists, replication should be stalled + verifyReplicationStuck(); + admin1.disableTable(DROPPED_TABLE); + // still stuck, source table still exists + verifyReplicationStuck(); + admin1.deleteTable(DROPPED_TABLE); + // now the source table is gone, replication should proceed, the + // offending edits be dropped + verifyReplicationProceeded(); + } + + private void verifyReplicationProceeded() throws Exception { + try (Table normalTable = utility1.getConnection().getTable(NORMAL_TABLE)) { + Put put = new Put(ROW); + put.addColumn(FAMILY, QUALIFIER, VALUE); + normalTable.put(put); + } + utility2.waitFor(NB_RETRIES * SLEEP_TIME, (Predicate) () -> { + try (Table normalTable = utility2.getConnection().getTable(NORMAL_TABLE)) { + Result result = normalTable.get(new Get(ROW).addColumn(FAMILY, QUALIFIER)); + return result != null && !result.isEmpty() + && Bytes.equals(VALUE, result.getValue(FAMILY, QUALIFIER)); + } + }); + } + + private void verifyReplicationStuck() throws Exception { + try (Table normalTable = utility1.getConnection().getTable(NORMAL_TABLE)) { + Put put = new Put(ROW); + put.addColumn(FAMILY, QUALIFIER, VALUE); + normalTable.put(put); + } + try (Table normalTable = utility2.getConnection().getTable(NORMAL_TABLE)) { + for (int i = 0; i < NB_RETRIES; i++) { + Result result = normalTable.get(new Get(ROW).addColumn(FAMILY, QUALIFIER)); + if (result != null && !result.isEmpty()) { + fail("Edit should have been stuck behind dropped tables, but value is " + Bytes + .toString(result.getValue(FAMILY, QUALIFIER))); + } else { + LOG.info("Row not replicated, let's wait a bit more..."); + Thread.sleep(SLEEP_TIME); + } + } + } + } +} \ No newline at end of file diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationStuckWithDeletedTableCFs.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationStuckWithDeletedTableCFs.java new file mode 100644 index 00000000000..c6491e51e5b --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationStuckWithDeletedTableCFs.java @@ -0,0 +1,183 @@ +/** + * 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; + +import static org.apache.hadoop.hbase.HConstants.REPLICATION_SCOPE_GLOBAL; +import static org.junit.Assert.fail; + +import java.util.Arrays; +import java.util.stream.Collectors; + +import org.apache.hadoop.conf.Configuration; +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.TableName; +import org.apache.hadoop.hbase.client.Admin; +import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; +import org.apache.hadoop.hbase.client.Get; +import org.apache.hadoop.hbase.client.Put; +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.testclassification.LargeTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Replication with dropped table will stuck as the default REPLICATION_DROP_ON_DELETED_TABLE_KEY + * is false. + */ +@Category({ LargeTests.class }) +public class TestReplicationStuckWithDeletedTableCFs { + + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestReplicationStuckWithDeletedTableCFs.class); + + private static final Logger LOG = + LoggerFactory.getLogger(TestReplicationStuckWithDeletedTableCFs.class); + + private static Configuration conf1 = HBaseConfiguration.create(); + private static Configuration conf2 = HBaseConfiguration.create(); + + protected static HBaseTestingUtility utility1; + protected static HBaseTestingUtility utility2; + + private static Admin admin1; + private static Admin admin2; + + private static final TableName TABLE = TableName.valueOf("normal-table"); + private static final byte[] ROW = Bytes.toBytes("row"); + private static final byte[] NORMAL_FAMILY = Bytes.toBytes("nf"); + private static final byte[] DROPPED_FAMILY = Bytes.toBytes("df"); + private static final byte[] QUALIFIER = Bytes.toBytes("q"); + private static final byte[] VALUE = Bytes.toBytes("value"); + + private static final String PEER_ID = "1"; + private static final long SLEEP_TIME = 1000; + private static final int NB_RETRIES = 10; + + @BeforeClass + public static void setUpBeforeClass() throws Exception { + conf1.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/1"); + conf1.setInt("replication.source.nb.capacity", 1); + utility1 = new HBaseTestingUtility(conf1); + utility1.startMiniZKCluster(); + MiniZooKeeperCluster miniZK = utility1.getZkCluster(); + conf1 = utility1.getConfiguration(); + + conf2 = HBaseConfiguration.create(conf1); + conf2.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/2"); + utility2 = new HBaseTestingUtility(conf2); + utility2.setZkCluster(miniZK); + + utility1.startMiniCluster(1); + utility2.startMiniCluster(1); + + admin1 = utility1.getAdmin(); + admin2 = utility2.getAdmin(); + } + + @AfterClass + public static void tearDownAfterClass() throws Exception { + utility2.shutdownMiniCluster(); + utility1.shutdownMiniCluster(); + } + + private void createTable(TableName tableName) throws Exception { + TableDescriptor desc = createTableDescriptor(DROPPED_FAMILY, NORMAL_FAMILY); + admin1.createTable(desc); + admin2.createTable(desc); + utility1.waitUntilAllRegionsAssigned(tableName); + utility2.waitUntilAllRegionsAssigned(tableName); + } + + @Test + public void testEditsStuckBehindDeletedCFs() throws Exception { + // add peer + ReplicationPeerConfig rpc = ReplicationPeerConfig.newBuilder() + .setClusterKey(utility2.getClusterKey()).setReplicateAllUserTables(true).build(); + admin1.addReplicationPeer(PEER_ID, rpc); + + // create table + createTable(TABLE); + + admin1.disableReplicationPeer(PEER_ID); + + try (Table droppedTable = utility1.getConnection().getTable(TABLE)) { + Put put = new Put(ROW); + put.addColumn(DROPPED_FAMILY, QUALIFIER, VALUE).addColumn(NORMAL_FAMILY, QUALIFIER, VALUE); + droppedTable.put(put); + } + + // delete cf + TableDescriptor desc = createTableDescriptor(NORMAL_FAMILY); + admin1.modifyTable(desc); + admin2.modifyTable(desc); + + admin1.enableReplicationPeer(PEER_ID); + + verifyReplicationStuck(); + + // Remove peer + admin1.removeReplicationPeer(PEER_ID); + // Drop table + admin1.disableTable(TABLE); + admin1.deleteTable(TABLE); + admin2.disableTable(TABLE); + admin2.deleteTable(TABLE); + } + + private void verifyReplicationStuck() throws Exception { + try (Table normalTable = utility1.getConnection().getTable(TABLE)) { + Put put = new Put(ROW); + put.addColumn(NORMAL_FAMILY, QUALIFIER, VALUE); + normalTable.put(put); + } + try (Table normalTable = utility2.getConnection().getTable(TABLE)) { + for (int i = 0; i < NB_RETRIES; i++) { + Result result = normalTable.get(new Get(ROW).addColumn(NORMAL_FAMILY, QUALIFIER)); + if (result != null && !result.isEmpty()) { + fail("Edit should have been stuck behind dropped tables, but value is " + Bytes + .toString(result.getValue(NORMAL_FAMILY, QUALIFIER))); + } else { + LOG.info("Row not replicated, let's wait a bit more..."); + Thread.sleep(SLEEP_TIME); + } + } + } + } + + private TableDescriptor createTableDescriptor(byte[]... cfs) { + return TableDescriptorBuilder.newBuilder(TABLE) + .setColumnFamilies(Arrays.stream(cfs).map(cf -> + ColumnFamilyDescriptorBuilder.newBuilder(cf).setScope(REPLICATION_SCOPE_GLOBAL).build()) + .collect(Collectors.toList()) + ).build(); + } +} \ No newline at end of file diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationStuckWithDroppedTable.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationStuckWithDroppedTable.java new file mode 100644 index 00000000000..cebf7d81781 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationStuckWithDroppedTable.java @@ -0,0 +1,176 @@ +/** + * 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; + +import static org.apache.hadoop.hbase.HConstants.REPLICATION_SCOPE_GLOBAL; +import static org.apache.hadoop.hbase.HConstants.ZOOKEEPER_ZNODE_PARENT; +import static org.junit.Assert.fail; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseClassTestRule; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HBaseTestingUtility; +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.Get; +import org.apache.hadoop.hbase.client.Put; +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.testclassification.LargeTests; +import org.apache.hadoop.hbase.util.Bytes; +import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Replication with dropped table will stuck as the default REPLICATION_DROP_ON_DELETED_TABLE_KEY + * is false. + */ +@Category({ LargeTests.class }) +public class TestReplicationStuckWithDroppedTable { + + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestReplicationStuckWithDroppedTable.class); + + private static final Logger LOG = + LoggerFactory.getLogger(TestReplicationEditsDroppedWithDroppedTable.class); + + private static Configuration conf1 = HBaseConfiguration.create(); + private static Configuration conf2 = HBaseConfiguration.create(); + + protected static HBaseTestingUtility utility1; + protected static HBaseTestingUtility utility2; + + private static Admin admin1; + private static Admin admin2; + + private static final TableName NORMAL_TABLE = TableName.valueOf("normal-table"); + private static final TableName DROPPED_TABLE = TableName.valueOf("dropped-table"); + private static final byte[] ROW = Bytes.toBytes("row"); + private static final byte[] FAMILY = Bytes.toBytes("f"); + private static final byte[] QUALIFIER = Bytes.toBytes("q"); + private static final byte[] VALUE = Bytes.toBytes("value"); + + private static final String PEER_ID = "1"; + private static final long SLEEP_TIME = 1000; + private static final int NB_RETRIES = 10; + + @BeforeClass + public static void setUpBeforeClass() throws Exception { + conf1.set(ZOOKEEPER_ZNODE_PARENT, "/1"); + conf1.setInt("replication.source.nb.capacity", 1); + utility1 = new HBaseTestingUtility(conf1); + utility1.startMiniZKCluster(); + MiniZooKeeperCluster miniZK = utility1.getZkCluster(); + conf1 = utility1.getConfiguration(); + + conf2 = HBaseConfiguration.create(conf1); + conf2.set(ZOOKEEPER_ZNODE_PARENT, "/2"); + utility2 = new HBaseTestingUtility(conf2); + utility2.setZkCluster(miniZK); + + utility1.startMiniCluster(1); + utility2.startMiniCluster(1); + + admin1 = utility1.getAdmin(); + admin2 = utility2.getAdmin(); + } + + @AfterClass + public static void tearDownAfterClass() throws Exception { + utility2.shutdownMiniCluster(); + utility1.shutdownMiniCluster(); + } + + private void createTable(TableName tableName) throws Exception { + TableDescriptor desc = TableDescriptorBuilder.newBuilder(tableName).setColumnFamily( + ColumnFamilyDescriptorBuilder.newBuilder(FAMILY).setScope(REPLICATION_SCOPE_GLOBAL).build() + ).build(); + admin1.createTable(desc); + admin2.createTable(desc); + utility1.waitUntilAllRegionsAssigned(tableName); + utility2.waitUntilAllRegionsAssigned(tableName); + } + + @Test + public void testEditsStuckBehindDroppedTable() throws Exception { + // add peer + ReplicationPeerConfig rpc = ReplicationPeerConfig.newBuilder() + .setClusterKey(utility2.getClusterKey()) + .setReplicateAllUserTables(true).build(); + admin1.addReplicationPeer(PEER_ID, rpc); + + // create table + createTable(NORMAL_TABLE); + createTable(DROPPED_TABLE); + + admin1.disableReplicationPeer(PEER_ID); + + try (Table droppedTable = utility1.getConnection().getTable(DROPPED_TABLE)) { + Put put = new Put(ROW); + put.addColumn(FAMILY, QUALIFIER, VALUE); + droppedTable.put(put); + } + + admin1.disableTable(DROPPED_TABLE); + admin1.deleteTable(DROPPED_TABLE); + admin2.disableTable(DROPPED_TABLE); + admin2.deleteTable(DROPPED_TABLE); + + admin1.enableReplicationPeer(PEER_ID); + + verifyReplicationStuck(); + + // Remove peer + admin1.removeReplicationPeer(PEER_ID); + // Drop table + admin1.disableTable(NORMAL_TABLE); + admin1.deleteTable(NORMAL_TABLE); + admin2.disableTable(NORMAL_TABLE); + admin2.deleteTable(NORMAL_TABLE); + } + + private void verifyReplicationStuck() throws Exception { + try (Table normalTable = utility1.getConnection().getTable(NORMAL_TABLE)) { + Put put = new Put(ROW); + put.addColumn(FAMILY, QUALIFIER, VALUE); + normalTable.put(put); + } + try (Table normalTable = utility2.getConnection().getTable(NORMAL_TABLE)) { + for (int i = 0; i < NB_RETRIES; i++) { + Result result = normalTable.get(new Get(ROW).addColumn(FAMILY, QUALIFIER)); + if (result != null && !result.isEmpty()) { + fail("Edit should have been stuck behind dropped tables, but value is " + Bytes + .toString(result.getValue(FAMILY, QUALIFIER))); + } else { + LOG.info("Row not replicated, let's wait a bit more..."); + Thread.sleep(SLEEP_TIME); + } + } + } + } +} \ No newline at end of file diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHBaseInterClusterReplicationEndpointFilterEdits.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHBaseInterClusterReplicationEndpointFilterEdits.java new file mode 100644 index 00000000000..2940b0d90fa --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestHBaseInterClusterReplicationEndpointFilterEdits.java @@ -0,0 +1,137 @@ +/** + * 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.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.List; + +import org.apache.hadoop.hbase.Cell; +import org.apache.hadoop.hbase.CellUtil; +import org.apache.hadoop.hbase.HBaseClassTestRule; +import org.apache.hadoop.hbase.HBaseTestingUtility; +import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.KeyValue.Type; +import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.replication.ReplicationEndpoint.Context; +import org.apache.hadoop.hbase.replication.ReplicationPeer; +import org.apache.hadoop.hbase.replication.ReplicationPeerConfig; +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.wal.WAL.Entry; +import org.apache.hadoop.hbase.wal.WALEdit; +import org.apache.hadoop.hbase.wal.WALKeyImpl; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +import org.apache.hbase.thirdparty.com.google.common.collect.Lists; + +/** + * Tests {@link HBaseInterClusterReplicationEndpoint#filterNotExistColumnFamilyEdits(List)} and + * {@link HBaseInterClusterReplicationEndpoint#filterNotExistTableEdits(List)} + */ +@Category({ ReplicationTests.class, MediumTests.class }) +public class TestHBaseInterClusterReplicationEndpointFilterEdits { + + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestHBaseInterClusterReplicationEndpointFilterEdits.class); + + private static final HBaseTestingUtility UTIL = new HBaseTestingUtility(); + + private static HBaseInterClusterReplicationEndpoint endpoint; + + private static final TableName TABLE1 = TableName.valueOf("T1"); + private static final TableName TABLE2 = TableName.valueOf("T2"); + + private static final byte[] FAMILY = Bytes.toBytes("CF"); + private static final byte[] NON_EXISTING_FAMILY = Bytes.toBytes("NECF"); + private static final byte[] QUALIFIER = Bytes.toBytes("Q"); + private static final byte[] ROW = Bytes.toBytes("r"); + private static final byte[] VALUE = Bytes.toBytes("v"); + + @BeforeClass + public static void setUpBeforeClass() throws Exception { + UTIL.startMiniCluster(); + ReplicationPeer replicationPeer = mock(ReplicationPeer.class); + ReplicationPeerConfig rpc = mock(ReplicationPeerConfig.class); + when(rpc.isSerial()).thenReturn(false); + when(replicationPeer.getPeerConfig()).thenReturn(rpc); + Context context = new Context(UTIL.getConfiguration(), UTIL.getConfiguration(), null, + null, null, replicationPeer, null, null, null); + endpoint = new HBaseInterClusterReplicationEndpoint(); + endpoint.init(context); + + UTIL.createTable(TABLE1, FAMILY); + } + + @AfterClass + public static void tearDownAfterClass() throws Exception { + UTIL.shutdownMiniCluster(); + } + + @Test + public void testFilterNotExistColumnFamilyEdits() { + List> entryList = new ArrayList<>(); + // should be filtered + Cell c1 = new KeyValue(ROW, NON_EXISTING_FAMILY, QUALIFIER, System.currentTimeMillis(), + Type.Put, VALUE); + Entry e1 = new Entry(new WALKeyImpl(new byte[32], TABLE1, System.currentTimeMillis()), + new WALEdit().add(c1)); + entryList.add(Lists.newArrayList(e1)); + // should be kept + Cell c2 = new KeyValue(ROW, FAMILY, QUALIFIER, System.currentTimeMillis(), Type.Put, VALUE); + Entry e2 = new Entry(new WALKeyImpl(new byte[32], TABLE1, System.currentTimeMillis()), + new WALEdit().add(c2)); + entryList.add(Lists.newArrayList(e2, e1)); + List> filtered = endpoint.filterNotExistColumnFamilyEdits(entryList); + assertEquals(1, filtered.size()); + assertEquals(1, filtered.get(0).get(0).getEdit().getCells().size()); + Cell cell = filtered.get(0).get(0).getEdit().getCells().get(0); + assertTrue(CellUtil.matchingFamily(cell, FAMILY)); + } + + @Test + public void testFilterNotExistTableEdits() { + List> entryList = new ArrayList<>(); + // should be filtered + Cell c1 = new KeyValue(ROW, FAMILY, QUALIFIER, System.currentTimeMillis(), Type.Put, VALUE); + Entry e1 = new Entry(new WALKeyImpl(new byte[32], TABLE2, System.currentTimeMillis()), + new WALEdit().add(c1)); + entryList.add(Lists.newArrayList(e1)); + // should be kept + Cell c2 = new KeyValue(ROW, FAMILY, QUALIFIER, System.currentTimeMillis(), Type.Put, VALUE); + Entry e2 = new Entry(new WALKeyImpl(new byte[32], TABLE1, System.currentTimeMillis()), + new WALEdit().add(c2)); + entryList.add(Lists.newArrayList(e2)); + List> filtered = endpoint.filterNotExistTableEdits(entryList); + assertEquals(1, filtered.size()); + Entry entry = filtered.get(0).get(0); + assertEquals(1, entry.getEdit().getCells().size()); + assertEquals(TABLE1, entry.getKey().getTableName()); + } + +}