001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.replication.regionserver; 019 020import static org.junit.Assert.assertEquals; 021import static org.junit.Assert.assertTrue; 022import static org.mockito.Mockito.mock; 023import static org.mockito.Mockito.when; 024 025import java.util.ArrayList; 026import java.util.List; 027import org.apache.hadoop.hbase.Cell; 028import org.apache.hadoop.hbase.CellUtil; 029import org.apache.hadoop.hbase.ExtendedCell; 030import org.apache.hadoop.hbase.HBaseClassTestRule; 031import org.apache.hadoop.hbase.HBaseTestingUtil; 032import org.apache.hadoop.hbase.KeyValue; 033import org.apache.hadoop.hbase.KeyValue.Type; 034import org.apache.hadoop.hbase.TableName; 035import org.apache.hadoop.hbase.replication.ReplicationEndpoint.Context; 036import org.apache.hadoop.hbase.replication.ReplicationPeer; 037import org.apache.hadoop.hbase.replication.ReplicationPeerConfig; 038import org.apache.hadoop.hbase.testclassification.MediumTests; 039import org.apache.hadoop.hbase.testclassification.ReplicationTests; 040import org.apache.hadoop.hbase.util.Bytes; 041import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; 042import org.apache.hadoop.hbase.wal.WAL.Entry; 043import org.apache.hadoop.hbase.wal.WALEdit; 044import org.apache.hadoop.hbase.wal.WALEditInternalHelper; 045import org.apache.hadoop.hbase.wal.WALKeyImpl; 046import org.junit.AfterClass; 047import org.junit.BeforeClass; 048import org.junit.ClassRule; 049import org.junit.Test; 050import org.junit.experimental.categories.Category; 051 052import org.apache.hbase.thirdparty.com.google.common.collect.Lists; 053 054/** 055 * Tests {@link HBaseInterClusterReplicationEndpoint#filterNotExistColumnFamilyEdits(List)} and 056 * {@link HBaseInterClusterReplicationEndpoint#filterNotExistTableEdits(List)} 057 */ 058@Category({ ReplicationTests.class, MediumTests.class }) 059public class TestHBaseInterClusterReplicationEndpointFilterEdits { 060 061 @ClassRule 062 public static final HBaseClassTestRule CLASS_RULE = 063 HBaseClassTestRule.forClass(TestHBaseInterClusterReplicationEndpointFilterEdits.class); 064 065 private static final HBaseTestingUtil UTIL = new HBaseTestingUtil(); 066 067 private static HBaseInterClusterReplicationEndpoint endpoint; 068 069 private static final TableName TABLE1 = TableName.valueOf("T1"); 070 private static final TableName TABLE2 = TableName.valueOf("T2"); 071 072 private static final byte[] FAMILY = Bytes.toBytes("CF"); 073 private static final byte[] NON_EXISTING_FAMILY = Bytes.toBytes("NECF"); 074 private static final byte[] QUALIFIER = Bytes.toBytes("Q"); 075 private static final byte[] ROW = Bytes.toBytes("r"); 076 private static final byte[] VALUE = Bytes.toBytes("v"); 077 078 @BeforeClass 079 public static void setUpBeforeClass() throws Exception { 080 UTIL.startMiniCluster(); 081 ReplicationPeer replicationPeer = mock(ReplicationPeer.class); 082 ReplicationPeerConfig rpc = mock(ReplicationPeerConfig.class); 083 when(rpc.isSerial()).thenReturn(false); 084 when(replicationPeer.getPeerConfig()).thenReturn(rpc); 085 when(rpc.getClusterKey()).thenReturn("hbase+zk://localhost:2181"); 086 Context context = new Context(null, UTIL.getConfiguration(), UTIL.getConfiguration(), null, 087 null, null, replicationPeer, null, null, null); 088 endpoint = new HBaseInterClusterReplicationEndpoint(); 089 endpoint.init(context); 090 091 UTIL.createTable(TABLE1, FAMILY); 092 } 093 094 @AfterClass 095 public static void tearDownAfterClass() throws Exception { 096 UTIL.shutdownMiniCluster(); 097 } 098 099 @Test 100 public void testFilterNotExistColumnFamilyEdits() { 101 List<List<Entry>> entryList = new ArrayList<>(); 102 // should be filtered 103 ExtendedCell c1 = new KeyValue(ROW, NON_EXISTING_FAMILY, QUALIFIER, 104 EnvironmentEdgeManager.currentTime(), Type.Put, VALUE); 105 Entry e1 = new Entry(new WALKeyImpl(new byte[32], TABLE1, EnvironmentEdgeManager.currentTime()), 106 WALEditInternalHelper.addExtendedCell(new WALEdit(), c1)); 107 entryList.add(Lists.newArrayList(e1)); 108 // should be kept 109 ExtendedCell c2 = 110 new KeyValue(ROW, FAMILY, QUALIFIER, EnvironmentEdgeManager.currentTime(), Type.Put, VALUE); 111 Entry e2 = new Entry(new WALKeyImpl(new byte[32], TABLE1, EnvironmentEdgeManager.currentTime()), 112 WALEditInternalHelper.addExtendedCell(new WALEdit(), c2)); 113 entryList.add(Lists.newArrayList(e2, e1)); 114 List<List<Entry>> filtered = endpoint.filterNotExistColumnFamilyEdits(entryList); 115 assertEquals(1, filtered.size()); 116 assertEquals(1, filtered.get(0).get(0).getEdit().getCells().size()); 117 Cell cell = filtered.get(0).get(0).getEdit().getCells().get(0); 118 assertTrue(CellUtil.matchingFamily(cell, FAMILY)); 119 } 120 121 @Test 122 public void testFilterNotExistTableEdits() { 123 List<List<Entry>> entryList = new ArrayList<>(); 124 // should be filtered 125 ExtendedCell c1 = 126 new KeyValue(ROW, FAMILY, QUALIFIER, EnvironmentEdgeManager.currentTime(), Type.Put, VALUE); 127 Entry e1 = new Entry(new WALKeyImpl(new byte[32], TABLE2, EnvironmentEdgeManager.currentTime()), 128 WALEditInternalHelper.addExtendedCell(new WALEdit(), c1)); 129 entryList.add(Lists.newArrayList(e1)); 130 // should be kept 131 ExtendedCell c2 = 132 new KeyValue(ROW, FAMILY, QUALIFIER, EnvironmentEdgeManager.currentTime(), Type.Put, VALUE); 133 Entry e2 = new Entry(new WALKeyImpl(new byte[32], TABLE1, EnvironmentEdgeManager.currentTime()), 134 WALEditInternalHelper.addExtendedCell(new WALEdit(), c2)); 135 entryList.add(Lists.newArrayList(e2)); 136 List<List<Entry>> filtered = endpoint.filterNotExistTableEdits(entryList); 137 assertEquals(1, filtered.size()); 138 Entry entry = filtered.get(0).get(0); 139 assertEquals(1, entry.getEdit().getCells().size()); 140 assertEquals(TABLE1, entry.getKey().getTableName()); 141 } 142 143}