HBASE-20294 Also cleanup last pushed sequence id in ReplicationBarrierCleaner
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/edf50495 Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/edf50495 Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/edf50495 Branch: refs/heads/HBASE-20388 Commit: edf50495022c798666f91e7adccb3b93e4200230 Parents: 7b7a222 Author: zhangduo <[email protected]> Authored: Sat Apr 14 14:56:18 2018 +0800 Committer: zhangduo <[email protected]> Committed: Sat Apr 14 14:56:18 2018 +0800 ---------------------------------------------------------------------- .../master/cleaner/ReplicationBarrierCleaner.java | 17 ++++++++++++----- .../cleaner/TestReplicationBarrierCleaner.java | 14 ++++++++++---- 2 files changed, 22 insertions(+), 9 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hbase/blob/edf50495/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBarrierCleaner.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBarrierCleaner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBarrierCleaner.java index 16b8fc5..ed631de 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBarrierCleaner.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/cleaner/ReplicationBarrierCleaner.java @@ -37,6 +37,7 @@ import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.master.replication.ReplicationPeerManager; import org.apache.hadoop.hbase.replication.ReplicationException; +import org.apache.hadoop.hbase.replication.ReplicationQueueStorage; import org.apache.yetus.audience.InterfaceAudience; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -75,6 +76,7 @@ public class ReplicationBarrierCleaner extends ScheduledChore { long cleanedRows = 0; long deletedRows = 0; long deletedBarriers = 0; + long deletedLastPushedSeqIds = 0; TableName tableName = null; List<String> peerIds = null; try (Table metaTable = conn.getTable(TableName.META_TABLE_NAME); @@ -123,11 +125,16 @@ public class ReplicationBarrierCleaner extends ScheduledChore { } else { index++; } - // A special case for merged/split region, where we are in the last closed range and the - // pushedSeqId is the last barrier minus 1. + // A special case for merged/split region, and also deleted tables, where we are in the last + // closed range and the pushedSeqId is the last barrier minus 1. if (index == barriers.length - 1 && pushedSeqId == barriers[barriers.length - 1] - 1) { // check if the region has already been removed, i.e, no catalog family if (!metaTable.exists(new Get(regionName).addFamily(HConstants.CATALOG_FAMILY))) { + ReplicationQueueStorage queueStorage = peerManager.getQueueStorage(); + for (String peerId: peerIds) { + queueStorage.removeLastSequenceIds(peerId, Arrays.asList(encodedRegionName)); + deletedLastPushedSeqIds++; + } metaTable .delete(new Delete(regionName).addFamily(HConstants.REPLICATION_BARRIER_FAMILY)); deletedRows++; @@ -153,9 +160,9 @@ public class ReplicationBarrierCleaner extends ScheduledChore { } if (totalRows > 0) { LOG.info( - "Cleanup replication barriers: " + - "totalRows {}, cleanedRows {}, deletedRows {}, deletedBarriers {}", - totalRows, cleanedRows, deletedRows, deletedBarriers); + "Cleanup replication barriers: totalRows {}, " + + "cleanedRows {}, deletedRows {}, deletedBarriers {}, deletedLastPushedSeqIds {}", + totalRows, cleanedRows, deletedRows, deletedBarriers, deletedLastPushedSeqIds); } } http://git-wip-us.apache.org/repos/asf/hbase/blob/edf50495/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBarrierCleaner.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBarrierCleaner.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBarrierCleaner.java index 671bc22..6e0d648 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBarrierCleaner.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/cleaner/TestReplicationBarrierCleaner.java @@ -20,6 +20,7 @@ package org.apache.hadoop.hbase.master.cleaner; import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertFalse; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -28,6 +29,7 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.io.IOException; +import java.util.Arrays; import java.util.Collections; import java.util.List; import org.apache.hadoop.hbase.HBaseClassTestRule; @@ -93,8 +95,8 @@ public class TestReplicationBarrierCleaner { @After public void tearDown() throws IOException { try (Table table = UTIL.getConnection().getTable(TableName.META_TABLE_NAME); - ResultScanner scanner = table.getScanner(new Scan().addFamily(HConstants.CATALOG_FAMILY) - .addFamily(HConstants.REPLICATION_BARRIER_FAMILY).setFilter(new FirstKeyOnlyFilter()))) { + ResultScanner scanner = table.getScanner(new Scan().addFamily(HConstants.CATALOG_FAMILY) + .addFamily(HConstants.REPLICATION_BARRIER_FAMILY).setFilter(new FirstKeyOnlyFilter()))) { for (;;) { Result result = scanner.next(); if (result == null) { @@ -144,7 +146,7 @@ public class TestReplicationBarrierCleaner { Put put = new Put(region.getRegionName(), EnvironmentEdgeManager.currentTime()); for (int i = 0; i < barriers.length; i++) { put.addColumn(HConstants.REPLICATION_BARRIER_FAMILY, HConstants.SEQNUM_QUALIFIER, - put.getTimeStamp() - barriers.length + i, Bytes.toBytes(barriers[i])); + put.getTimestamp() - barriers.length + i, Bytes.toBytes(barriers[i])); } try (Table table = UTIL.getConnection().getTable(TableName.META_TABLE_NAME)) { table.put(put); @@ -260,15 +262,17 @@ public class TestReplicationBarrierCleaner { addBarrier(region, 40, 50, 60); fillCatalogFamily(region); + String peerId = "1"; ReplicationQueueStorage queueStorage = create(59L); @SuppressWarnings("unchecked") - ReplicationPeerManager peerManager = create(queueStorage, Lists.newArrayList("1")); + ReplicationPeerManager peerManager = create(queueStorage, Lists.newArrayList(peerId)); ReplicationBarrierCleaner cleaner = create(peerManager); // we have something in catalog family, so only delete 40 cleaner.chore(); assertArrayEquals(new long[] { 50, 60 }, MetaTableAccessor.getReplicationBarrier(UTIL.getConnection(), region.getRegionName())); + verify(queueStorage, never()).removeLastSequenceIds(anyString(), anyList()); // No catalog family, then we should remove the whole row clearCatalogFamily(region); @@ -277,6 +281,8 @@ public class TestReplicationBarrierCleaner { assertFalse(table .exists(new Get(region.getRegionName()).addFamily(HConstants.REPLICATION_BARRIER_FAMILY))); } + verify(queueStorage, times(1)).removeLastSequenceIds(peerId, + Arrays.asList(region.getEncodedName())); } private static class WarnOnlyStoppable implements Stoppable {
