From dd6f4525e7a9eaceb6ed4f97059b2dd3c532d323 Mon Sep 17 00:00:00 2001 From: zhangduo Date: Fri, 9 Mar 2018 15:00:59 +0800 Subject: [PATCH] HBASE-20148 Make serial replication as a option for a peer instead of a table --- .../apache/hadoop/hbase/HTableDescriptor.java | 8 ----- .../hadoop/hbase/client/TableDescriptor.java | 20 ++++++----- .../hbase/client/TableDescriptorBuilder.java | 9 ----- .../ReplicationPeerConfigUtil.java | 5 +++ .../replication/ReplicationPeerConfig.java | 32 ++++++++++++----- .../ReplicationPeerConfigBuilder.java | 12 +++++++ .../org/apache/hadoop/hbase/HConstants.java | 6 ---- .../src/main/protobuf/Replication.proto | 1 + .../hbase/replication/ReplicationUtils.java | 3 ++ .../master/assignment/RegionStateStore.java | 14 ++++---- .../replication/ScopeWALEntryFilter.java | 34 +++++++++---------- .../regionserver/ReplicationSource.java | 4 +++ .../ReplicationSourceWALActionListener.java | 10 +----- .../ReplicationSourceWALReader.java | 6 ++-- .../SerialReplicationChecker.java | 2 +- .../apache/hadoop/hbase/wal/WALKeyImpl.java | 8 ----- .../TestReplicationWALEntryFilters.java | 15 +++++--- .../replication/TestSerialReplication.java | 9 ++--- 18 files changed, 105 insertions(+), 93 deletions(-) diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/HTableDescriptor.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/HTableDescriptor.java index 3652d104f93..db8870dad12 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/HTableDescriptor.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/HTableDescriptor.java @@ -536,14 +536,6 @@ public class HTableDescriptor implements TableDescriptor, Comparable cf.getScope() == HConstants.REPLICATION_SCOPE_GLOBAL); + } + /** * Check if the table's cfs' replication scope matched with the replication state * @param enabled replication state @@ -284,8 +289,7 @@ public interface TableDescriptor { boolean hasDisabled = false; for (ColumnFamilyDescriptor cf : getColumnFamilies()) { - if (cf.getScope() != HConstants.REPLICATION_SCOPE_GLOBAL && - cf.getScope() != HConstants.REPLICATION_SCOPE_SERIAL) { + if (cf.getScope() != HConstants.REPLICATION_SCOPE_GLOBAL) { hasDisabled = true; } else { hasEnabled = true; diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java index 0855f87f6b0..c1db64bf4c4 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/TableDescriptorBuilder.java @@ -1053,15 +1053,6 @@ public class TableDescriptorBuilder { return families.values().toArray(new ColumnFamilyDescriptor[families.size()]); } - /** - * Return true if there are at least one cf whose replication scope is serial. - */ - @Override - public boolean hasSerialReplicationScope() { - return families.values().stream() - .anyMatch(column -> column.getScope() == HConstants.REPLICATION_SCOPE_SERIAL); - } - /** * Returns the configured replicas per region */ diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationPeerConfigUtil.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationPeerConfigUtil.java index a234a9b8010..b1c17136dc1 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationPeerConfigUtil.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationPeerConfigUtil.java @@ -303,6 +303,10 @@ public final class ReplicationPeerConfigUtil { builder.setReplicateAllUserTables(peer.getReplicateAll()); } + if (peer.hasSerial()) { + builder.setSerial(peer.getSerial()); + } + Map> excludeTableCFsMap = convert2Map(peer.getExcludeTableCfsList() .toArray(new ReplicationProtos.TableCF[peer.getExcludeTableCfsCount()])); if (excludeTableCFsMap != null) { @@ -357,6 +361,7 @@ public final class ReplicationPeerConfigUtil { builder.setBandwidth(peerConfig.getBandwidth()); builder.setReplicateAll(peerConfig.replicateAllUserTables()); + builder.setSerial(peerConfig.isSerial()); ReplicationProtos.TableCF[] excludeTableCFs = convert(peerConfig.getExcludeTableCFsMap()); if (excludeTableCFs != null) { diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfig.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfig.java index bf8d0303597..e0d9a4c86b7 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfig.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfig.java @@ -46,6 +46,7 @@ public class ReplicationPeerConfig { private Map> excludeTableCFsMap = null; private Set excludeNamespaces = null; private long bandwidth = 0; + private final boolean serial; private ReplicationPeerConfig(ReplicationPeerConfigBuilderImpl builder) { this.clusterKey = builder.clusterKey; @@ -64,6 +65,7 @@ public class ReplicationPeerConfig { builder.excludeNamespaces != null ? Collections.unmodifiableSet(builder.excludeNamespaces) : null; this.bandwidth = builder.bandwidth; + this.serial = builder.serial; } private Map> @@ -82,6 +84,7 @@ public class ReplicationPeerConfig { public ReplicationPeerConfig() { this.peerData = new TreeMap<>(Bytes.BYTES_COMPARATOR); this.configuration = new HashMap<>(0); + this.serial = false; } /** @@ -214,16 +217,20 @@ public class ReplicationPeerConfig { return new ReplicationPeerConfigBuilderImpl(); } + public boolean isSerial() { + return serial; + } + public static ReplicationPeerConfigBuilder newBuilder(ReplicationPeerConfig peerConfig) { ReplicationPeerConfigBuilderImpl builder = new ReplicationPeerConfigBuilderImpl(); builder.setClusterKey(peerConfig.getClusterKey()) - .setReplicationEndpointImpl(peerConfig.getReplicationEndpointImpl()) - .putAllPeerData(peerConfig.getPeerData()).putAllConfiguration(peerConfig.getConfiguration()) - .setTableCFsMap(peerConfig.getTableCFsMap()).setNamespaces(peerConfig.getNamespaces()) - .setReplicateAllUserTables(peerConfig.replicateAllUserTables()) - .setExcludeTableCFsMap(peerConfig.getExcludeTableCFsMap()) - .setExcludeNamespaces(peerConfig.getExcludeNamespaces()) - .setBandwidth(peerConfig.getBandwidth()); + .setReplicationEndpointImpl(peerConfig.getReplicationEndpointImpl()) + .putAllPeerData(peerConfig.getPeerData()).putAllConfiguration(peerConfig.getConfiguration()) + .setTableCFsMap(peerConfig.getTableCFsMap()).setNamespaces(peerConfig.getNamespaces()) + .setReplicateAllUserTables(peerConfig.replicateAllUserTables()) + .setExcludeTableCFsMap(peerConfig.getExcludeTableCFsMap()) + .setExcludeNamespaces(peerConfig.getExcludeNamespaces()) + .setBandwidth(peerConfig.getBandwidth()).setSerial(peerConfig.isSerial()); return builder; } @@ -250,6 +257,8 @@ public class ReplicationPeerConfig { private long bandwidth = 0; + private boolean serial = false; + @Override public ReplicationPeerConfigBuilder setClusterKey(String clusterKey) { this.clusterKey = clusterKey; @@ -312,6 +321,12 @@ public class ReplicationPeerConfig { return this; } + @Override + public ReplicationPeerConfigBuilder setSerial(boolean serial) { + this.serial = serial; + return this; + } + @Override public ReplicationPeerConfig build() { // It would be nice to validate the configuration, but we have to work with "old" data @@ -340,7 +355,8 @@ public class ReplicationPeerConfig { builder.append("tableCFs=").append(tableCFsMap.toString()).append(","); } } - builder.append("bandwidth=").append(bandwidth); + builder.append("bandwidth=").append(bandwidth).append(","); + builder.append("serial=").append(serial); return builder.toString(); } diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfigBuilder.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfigBuilder.java index 0b2f2e2989d..4c531c5ada9 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfigBuilder.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/replication/ReplicationPeerConfigBuilder.java @@ -137,6 +137,18 @@ public interface ReplicationPeerConfigBuilder { */ ReplicationPeerConfigBuilder setExcludeNamespaces(Set namespaces); + /** + *

+ * Sets whether we should preserve order when replicating, i.e, serial replication. + *

+ *

+ * Default {@code false}. + *

+ * @param serial {@code true} means preserve order, otherwise {@code false}. + * @return {@code this} + */ + ReplicationPeerConfigBuilder setSerial(boolean serial); + /** * Builds the configuration object from the current state of {@code this}. * @return A {@link ReplicationPeerConfig} instance. 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 3dd0ac83f9f..0039a56d429 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 @@ -646,12 +646,6 @@ public final class HConstants { */ public static final int REPLICATION_SCOPE_GLOBAL = 1; - /** - * Scope tag for serially scoped data - * This data will be replicated to all peers by the order of sequence id. - */ - public static final int REPLICATION_SCOPE_SERIAL = 2; - /** * Default cluster ID, cannot be used to identify a cluster so a key with * this value means it wasn't meant for replication. diff --git a/hbase-protocol-shaded/src/main/protobuf/Replication.proto b/hbase-protocol-shaded/src/main/protobuf/Replication.proto index 9f7b4c257e3..557b87cf74e 100644 --- a/hbase-protocol-shaded/src/main/protobuf/Replication.proto +++ b/hbase-protocol-shaded/src/main/protobuf/Replication.proto @@ -48,6 +48,7 @@ message ReplicationPeer { optional bool replicate_all = 8; repeated TableCF exclude_table_cfs = 9; repeated bytes exclude_namespaces = 10; + optional bool serial = 11; } /** diff --git a/hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationUtils.java b/hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationUtils.java index 11507aa429e..2a6870ae76d 100644 --- a/hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationUtils.java +++ b/hbase-replication/src/main/java/org/apache/hadoop/hbase/replication/ReplicationUtils.java @@ -115,6 +115,9 @@ public final class ReplicationUtils { if (rpc1.replicateAllUserTables() != rpc2.replicateAllUserTables()) { return false; } + if (rpc1.isSerial() != rpc2.isSerial()) { + return false; + } if (rpc1.replicateAllUserTables()) { return isNamespacesEqual(rpc1.getExcludeNamespaces(), rpc2.getExcludeNamespaces()) && isTableCFsEqual(rpc1.getExcludeTableCFsMap(), rpc2.getExcludeTableCFsMap()); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionStateStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionStateStore.java index 1ffc31f0848..c80172020bf 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionStateStore.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionStateStore.java @@ -165,7 +165,7 @@ public class RegionStateStore { MetaTableAccessor.addLocation(put, regionLocation, openSeqNum, replicaId); // only update replication barrier for default replica if (regionInfo.getReplicaId() == RegionInfo.DEFAULT_REPLICA_ID && - hasSerialReplicationScope(regionInfo.getTable())) { + hasGlobalReplicationScope(regionInfo.getTable())) { MetaTableAccessor.addReplicationBarrier(put, openSeqNum); } info.append(", openSeqNum=").append(openSeqNum); @@ -224,7 +224,7 @@ public class RegionStateStore { ServerName serverName) throws IOException { TableDescriptor htd = getTableDescriptor(parent.getTable()); long parentOpenSeqNum = HConstants.NO_SEQNUM; - if (htd.hasSerialReplicationScope()) { + if (htd.hasGlobalReplicationScope()) { parentOpenSeqNum = getOpenSeqNumForParentRegion(parent); } MetaTableAccessor.splitRegion(master.getConnection(), parent, parentOpenSeqNum, hriA, hriB, @@ -239,7 +239,7 @@ public class RegionStateStore { TableDescriptor htd = getTableDescriptor(child.getTable()); long regionAOpenSeqNum = -1L; long regionBOpenSeqNum = -1L; - if (htd.hasSerialReplicationScope()) { + if (htd.hasGlobalReplicationScope()) { regionAOpenSeqNum = getOpenSeqNumForParentRegion(hriA); regionBOpenSeqNum = getOpenSeqNumForParentRegion(hriB); } @@ -261,12 +261,12 @@ public class RegionStateStore { // ========================================================================== // Table Descriptors helpers // ========================================================================== - private boolean hasSerialReplicationScope(TableName tableName) throws IOException { - return hasSerialReplicationScope(getTableDescriptor(tableName)); + private boolean hasGlobalReplicationScope(TableName tableName) throws IOException { + return hasGlobalReplicationScope(getTableDescriptor(tableName)); } - private boolean hasSerialReplicationScope(TableDescriptor htd) { - return htd != null ? htd.hasSerialReplicationScope() : false; + private boolean hasGlobalReplicationScope(TableDescriptor htd) { + return htd != null ? htd.hasGlobalReplicationScope() : false; } private int getRegionReplication(TableDescriptor htd) { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ScopeWALEntryFilter.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ScopeWALEntryFilter.java index 6a2fbcf429b..f8722eb3da4 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ScopeWALEntryFilter.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/ScopeWALEntryFilter.java @@ -37,31 +37,31 @@ public class ScopeWALEntryFilter implements WALEntryFilter, WALCellFilter { @Override public Entry filter(Entry entry) { + // Do not filter out an entire entry by replication scopes. As now we support serial + // replication, the sequence id of a marker is also needed by upper layer. We will filter out + // all the cells in the filterCell method below if the replication scopes is null or empty. + return entry; + } + + private boolean hasGlobalScope(NavigableMap scopes, byte[] family) { + Integer scope = scopes.get(family); + return scope != null && scope.intValue() == HConstants.REPLICATION_SCOPE_GLOBAL; + } + @Override + public Cell filterCell(Entry entry, Cell cell) { NavigableMap scopes = entry.getKey().getReplicationScopes(); if (scopes == null || scopes.isEmpty()) { return null; } - return entry; - } - - @Override - public Cell filterCell(Entry entry, Cell cell) { - final NavigableMap scopes = entry.getKey().getReplicationScopes(); - // The scope will be null or empty if - // there's nothing to replicate in that WALEdit - byte[] fam = CellUtil.cloneFamily(cell); + byte[] family = CellUtil.cloneFamily(cell); if (CellUtil.matchingColumn(cell, WALEdit.METAFAMILY, WALEdit.BULK_LOAD)) { - cell = bulkLoadFilter.filterCell(cell, new Predicate() { + return bulkLoadFilter.filterCell(cell, new Predicate() { @Override - public boolean apply(byte[] fam) { - return !scopes.containsKey(fam) || scopes.get(fam) == HConstants.REPLICATION_SCOPE_LOCAL; + public boolean apply(byte[] family) { + return !hasGlobalScope(scopes, family); } }); - } else { - if (!scopes.containsKey(fam) || scopes.get(fam) == HConstants.REPLICATION_SCOPE_LOCAL) { - return null; - } } - return cell; + return hasGlobalScope(scopes, family) ? cell : null; } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java index 86e7f98b943..2f9cd56bbb8 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java @@ -382,6 +382,10 @@ public class ReplicationSource implements ReplicationSourceInterface { return replicationPeer.isPeerEnabled(); } + public boolean isSerial() { + return replicationPeer.getPeerConfig().isSerial(); + } + private void initialize() { int sleepMultiplier = 1; while (this.isSourceActive()) { diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALActionListener.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALActionListener.java index 95fc6a088a9..27b25c45a75 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALActionListener.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALActionListener.java @@ -72,18 +72,10 @@ class ReplicationSourceWALActionListener implements WALActionsListener { if (ReplicationUtils.isReplicationForBulkLoadDataEnabled(conf)) { return; } - WALKeyImpl keyImpl = (WALKeyImpl) logKey; - // For serial replication we need to count all the sequence ids even for markers, so here we - // always need to retain the replication scopes to let the replication wal reader to know that - // we need serial replication. The ScopeWALEntryFilter will help filtering out the cell for - // WALEdit.METAFAMILY. - if (keyImpl.hasSerialReplicationScope()) { - return; - } // For replay, or if all the cells are markers, do not need to store replication scope. if (logEdit.isReplay() || logEdit.getCells().stream().allMatch(c -> CellUtil.matchingFamily(c, WALEdit.METAFAMILY))) { - keyImpl.clearReplicationScope(); + ((WALKeyImpl) logKey).clearReplicationScope(); } } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java index ad3baafa15f..da92a09edbe 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java @@ -186,9 +186,9 @@ public class ReplicationSourceWALReader extends Thread { new WALEntryBatch(replicationBatchCountCapacity, entryStream.getCurrentPath()); do { Entry entry = entryStream.peek(); - boolean hasSerialReplicationScope = entry.getKey().hasSerialReplicationScope(); + boolean isSerial = source.isSerial(); boolean doFiltering = true; - if (hasSerialReplicationScope) { + if (isSerial) { if (firstCellInEntryBeforeFiltering == null) { assert !entry.getEdit().isEmpty() : "should not write empty edits"; // Used to locate the region record in meta table. In WAL we only have the table name and @@ -208,7 +208,7 @@ public class ReplicationSourceWALReader extends Thread { entry = filterEntry(entry); } if (entry != null) { - if (hasSerialReplicationScope) { + if (isSerial) { if (!serialReplicationChecker.canPush(entry, firstCellInEntryBeforeFiltering)) { if (batch.getLastWalPosition() > positionBefore) { // we have something that can push, break diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/SerialReplicationChecker.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/SerialReplicationChecker.java index 9276359172d..b775d25cb1d 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/SerialReplicationChecker.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/SerialReplicationChecker.java @@ -266,7 +266,7 @@ class SerialReplicationChecker { throws IOException, InterruptedException { byte[] row = CellUtil.cloneRow(firstCellInEdit); while (!canPush(entry, row)) { - LOG.debug("Can not push{}, wait", entry); + LOG.debug("Can not push {}, wait", entry); Thread.sleep(waitTimeMs); } } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALKeyImpl.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALKeyImpl.java index ac23d1d952a..8828239f48d 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALKeyImpl.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALKeyImpl.java @@ -419,14 +419,6 @@ public class WALKeyImpl implements WALKey { setReplicationScope(null); } - public boolean hasSerialReplicationScope() { - if (replicationScope == null || replicationScope.isEmpty()) { - return false; - } - return replicationScope.values().stream() - .anyMatch(scope -> scope.intValue() == HConstants.REPLICATION_SCOPE_SERIAL); - } - /** * Marks that the cluster with the given clusterId has consumed the change */ diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationWALEntryFilters.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationWALEntryFilters.java index 67a25512534..f2c5e50493a 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationWALEntryFilters.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationWALEntryFilters.java @@ -18,6 +18,7 @@ package org.apache.hadoop.hbase.replication; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -32,9 +33,9 @@ import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellComparatorImpl; import org.apache.hadoop.hbase.HBaseClassTestRule; import org.apache.hadoop.hbase.HConstants; -import org.apache.hadoop.hbase.HRegionInfo; import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.TableName; +import org.apache.hadoop.hbase.client.RegionInfoBuilder; import org.apache.hadoop.hbase.testclassification.ReplicationTests; import org.apache.hadoop.hbase.testclassification.SmallTests; import org.apache.hadoop.hbase.util.Bytes; @@ -48,7 +49,7 @@ import org.junit.experimental.categories.Category; import org.apache.hbase.thirdparty.com.google.common.collect.Lists; -@Category({ReplicationTests.class, SmallTests.class}) +@Category({ ReplicationTests.class, SmallTests.class }) public class TestReplicationWALEntryFilters { @ClassRule @@ -65,7 +66,8 @@ public class TestReplicationWALEntryFilters { SystemTableWALEntryFilter filter = new SystemTableWALEntryFilter(); // meta - WALKeyImpl key1 = new WALKeyImpl(HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes(), + WALKeyImpl key1 = + new WALKeyImpl(RegionInfoBuilder.FIRST_META_REGIONINFO.getEncodedNameAsBytes(), TableName.META_TABLE_NAME, System.currentTimeMillis()); Entry metaEntry = new Entry(key1, null); @@ -96,12 +98,15 @@ public class TestReplicationWALEntryFilters { Entry userEntryEmpty = createEntry(null); // no scopes - assertEquals(null, filter.filter(userEntry)); + // now we will not filter out entries without a replication scope since serial replication still + // need the sequence id, but the cells will all be filtered out. + assertTrue(filter.filter(userEntry).getEdit().isEmpty()); // empty scopes + // ditto TreeMap scopes = new TreeMap<>(Bytes.BYTES_COMPARATOR); userEntry = createEntry(scopes, a, b); - assertEquals(null, filter.filter(userEntry)); + assertTrue(filter.filter(userEntry).getEdit().isEmpty()); // different scope scopes = new TreeMap<>(Bytes.BYTES_COMPARATOR); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestSerialReplication.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestSerialReplication.java index bf6c0c8fe99..f8efcf07822 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestSerialReplication.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestSerialReplication.java @@ -156,7 +156,8 @@ public class TestSerialReplication { // add in disable state, so later when enabling it all sources will start push together. UTIL.getAdmin().addReplicationPeer(PEER_ID, ReplicationPeerConfig.newBuilder().setClusterKey("127.0.0.1:2181:/hbase") - .setReplicationEndpointImpl(LocalReplicationEndpoint.class.getName()).build(), + .setReplicationEndpointImpl(LocalReplicationEndpoint.class.getName()).setSerial(true) + .build(), false); } @@ -234,7 +235,7 @@ public class TestSerialReplication { TableName tableName = TableName.valueOf(name.getMethodName()); UTIL.getAdmin().createTable( TableDescriptorBuilder.newBuilder(tableName).addColumnFamily(ColumnFamilyDescriptorBuilder - .newBuilder(CF).setScope(HConstants.REPLICATION_SCOPE_SERIAL).build()).build()); + .newBuilder(CF).setScope(HConstants.REPLICATION_SCOPE_GLOBAL).build()).build()); UTIL.waitTableAvailable(tableName); try (Table table = UTIL.getConnection().getTable(tableName)) { for (int i = 0; i < 100; i++) { @@ -273,7 +274,7 @@ public class TestSerialReplication { TableName tableName = TableName.valueOf(name.getMethodName()); UTIL.getAdmin().createTable( TableDescriptorBuilder.newBuilder(tableName).addColumnFamily(ColumnFamilyDescriptorBuilder - .newBuilder(CF).setScope(HConstants.REPLICATION_SCOPE_SERIAL).build()).build()); + .newBuilder(CF).setScope(HConstants.REPLICATION_SCOPE_GLOBAL).build()).build()); UTIL.waitTableAvailable(tableName); try (Table table = UTIL.getConnection().getTable(tableName)) { for (int i = 0; i < 100; i++) { @@ -330,7 +331,7 @@ public class TestSerialReplication { UTIL.getAdmin().createTable( TableDescriptorBuilder.newBuilder(tableName) .addColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(CF) - .setScope(HConstants.REPLICATION_SCOPE_SERIAL).build()) + .setScope(HConstants.REPLICATION_SCOPE_GLOBAL).build()) .build(), new byte[][] { splitKey }); UTIL.waitTableAvailable(tableName);