Source code

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;
019
020import static org.junit.Assert.assertArrayEquals;
021import static org.junit.Assert.fail;
022
023import java.util.ArrayList;
024import java.util.List;
025import org.apache.hadoop.hbase.HBaseClassTestRule;
026import org.apache.hadoop.hbase.HColumnDescriptor;
027import org.apache.hadoop.hbase.HConstants;
028import org.apache.hadoop.hbase.HTableDescriptor;
029import org.apache.hadoop.hbase.NamespaceDescriptor;
030import org.apache.hadoop.hbase.TableName;
031import org.apache.hadoop.hbase.client.Admin;
032import org.apache.hadoop.hbase.client.Connection;
033import org.apache.hadoop.hbase.client.ConnectionFactory;
034import org.apache.hadoop.hbase.client.Get;
035import org.apache.hadoop.hbase.client.Put;
036import org.apache.hadoop.hbase.client.Result;
037import org.apache.hadoop.hbase.client.ResultScanner;
038import org.apache.hadoop.hbase.client.Scan;
039import org.apache.hadoop.hbase.client.Table;
040import org.apache.hadoop.hbase.testclassification.LargeTests;
041import org.apache.hadoop.hbase.util.Bytes;
042import org.apache.hadoop.hbase.util.JVMClusterUtil;
043import org.junit.Before;
044import org.junit.ClassRule;
045import org.junit.Test;
046import org.junit.experimental.categories.Category;
047import org.slf4j.Logger;
048import org.slf4j.LoggerFactory;
049
050@Category(LargeTests.class)
051public class TestReplicationDroppedTables extends TestReplicationBase {
052
053  @ClassRule
054  public static final HBaseClassTestRule CLASS_RULE =
055      HBaseClassTestRule.forClass(TestReplicationDroppedTables.class);
056
057  private static final Logger LOG = LoggerFactory.getLogger(TestReplicationDroppedTables.class);
058
059  /**
060   * @throws java.lang.Exception
061   */
062  @Before
063  public void setUp() throws Exception {
064    // Starting and stopping replication can make us miss new logs,
065    // rolling like this makes sure the most recent one gets added to the queue
066    for ( JVMClusterUtil.RegionServerThread r :
067        utility1.getHBaseCluster().getRegionServerThreads()) {
068      utility1.getAdmin().rollWALWriter(r.getRegionServer().getServerName());
069    }
070    int rowCount = utility1.countRows(tableName);
071    utility1.deleteTableData(tableName);
072    // truncating the table will send one Delete per row to the slave cluster
073    // in an async fashion, which is why we cannot just call deleteTableData on
074    // utility2 since late writes could make it to the slave in some way.
075    // Instead, we truncate the first table and wait for all the Deletes to
076    // make it to the slave.
077    Scan scan = new Scan();
078    int lastCount = 0;
079    for (int i = 0; i < NB_RETRIES; i++) {
080      if (i==NB_RETRIES-1) {
081        fail("Waited too much time for truncate");
082      }
083      ResultScanner scanner = htable2.getScanner(scan);
084      Result[] res = scanner.next(rowCount);
085      scanner.close();
086      if (res.length != 0) {
087        if (res.length < lastCount) {
088          i--; // Don't increment timeout if we make progress
089        }
090        lastCount = res.length;
091        LOG.info("Still got " + res.length + " rows");
092        Thread.sleep(SLEEP_TIME);
093      } else {
094        break;
095      }
096    }
097  }
098
099  @Test
100  public void testEditsStuckBehindDroppedTable() throws Exception {
101    // Sanity check
102    // Make sure by default edits for dropped tables stall the replication queue, even when the
103    // table(s) in question have been deleted on both ends.
104    testEditsBehindDroppedTable(false, "test_dropped");
105  }
106
107  @Test
108  public void testEditsDroppedWithDroppedTable() throws Exception {
109    // Make sure by default edits for dropped tables are themselves dropped when the
110    // table(s) in question have been deleted on both ends.
111    testEditsBehindDroppedTable(true, "test_dropped");
112  }
113
114  @Test
115  public void testEditsDroppedWithDroppedTableNS() throws Exception {
116    // also try with a namespace
117    Connection connection1 = ConnectionFactory.createConnection(conf1);
118    try (Admin admin1 = connection1.getAdmin()) {
119      admin1.createNamespace(NamespaceDescriptor.create("NS").build());
120    }
121    Connection connection2 = ConnectionFactory.createConnection(conf2);
122    try (Admin admin2 = connection2.getAdmin()) {
123      admin2.createNamespace(NamespaceDescriptor.create("NS").build());
124    }
125    testEditsBehindDroppedTable(true, "NS:test_dropped");
126  }
127
128  private void testEditsBehindDroppedTable(boolean allowProceeding, String tName) throws Exception {
129    conf1.setBoolean(HConstants.REPLICATION_DROP_ON_DELETED_TABLE_KEY, allowProceeding);
130    conf1.setInt(HConstants.REPLICATION_SOURCE_MAXTHREADS_KEY, 1);
131
132    // make sure we have a single region server only, so that all
133    // edits for all tables go there
134    utility1.shutdownMiniHBaseCluster();
135    utility1.startMiniHBaseCluster(1, 1);
136
137    TableName tablename = TableName.valueOf(tName);
138    byte[] familyname = Bytes.toBytes("fam");
139    byte[] row = Bytes.toBytes("row");
140
141    HTableDescriptor table = new HTableDescriptor(tablename);
142    HColumnDescriptor fam = new HColumnDescriptor(familyname);
143    fam.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
144    table.addFamily(fam);
145
146    Connection connection1 = ConnectionFactory.createConnection(conf1);
147    Connection connection2 = ConnectionFactory.createConnection(conf2);
148    try (Admin admin1 = connection1.getAdmin()) {
149      admin1.createTable(table);
150    }
151    try (Admin admin2 = connection2.getAdmin()) {
152      admin2.createTable(table);
153    }
154    utility1.waitUntilAllRegionsAssigned(tablename);
155    utility2.waitUntilAllRegionsAssigned(tablename);
156
157    Table lHtable1 = utility1.getConnection().getTable(tablename);
158
159    // now suspend replication
160    admin.disablePeer("2");
161
162    // put some data (lead with 0 so the edit gets sorted before the other table's edits
163    //   in the replication batch)
164    // write a bunch of edits, making sure we fill a batch
165    byte[] rowkey = Bytes.toBytes(0+" put on table to be dropped");
166    Put put = new Put(rowkey);
167    put.addColumn(familyname, row, row);
168    lHtable1.put(put);
169
170    rowkey = Bytes.toBytes("normal put");
171    put = new Put(rowkey);
172    put.addColumn(famName, row, row);
173    htable1.put(put);
174
175    try (Admin admin1 = connection1.getAdmin()) {
176      admin1.disableTable(tablename);
177      admin1.deleteTable(tablename);
178    }
179    try (Admin admin2 = connection2.getAdmin()) {
180      admin2.disableTable(tablename);
181      admin2.deleteTable(tablename);
182    }
183
184    admin.enablePeer("2");
185    if (allowProceeding) {
186      // in this we'd expect the key to make it over
187      verifyReplicationProceeded(rowkey);
188    } else {
189      verifyReplicationStuck(rowkey);
190    }
191    // just to be safe
192    conf1.setBoolean(HConstants.REPLICATION_DROP_ON_DELETED_TABLE_KEY, false);
193  }
194
195  @Test
196  public void testEditsBehindDroppedTableTiming() throws Exception {
197    conf1.setBoolean(HConstants.REPLICATION_DROP_ON_DELETED_TABLE_KEY, true);
198    conf1.setInt(HConstants.REPLICATION_SOURCE_MAXTHREADS_KEY, 1);
199
200    // make sure we have a single region server only, so that all
201    // edits for all tables go there
202    utility1.shutdownMiniHBaseCluster();
203    utility1.startMiniHBaseCluster(1, 1);
204
205    TableName tablename = TableName.valueOf("testdroppedtimed");
206    byte[] familyname = Bytes.toBytes("fam");
207    byte[] row = Bytes.toBytes("row");
208
209    HTableDescriptor table = new HTableDescriptor(tablename);
210    HColumnDescriptor fam = new HColumnDescriptor(familyname);
211    fam.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
212    table.addFamily(fam);
213
214    Connection connection1 = ConnectionFactory.createConnection(conf1);
215    Connection connection2 = ConnectionFactory.createConnection(conf2);
216    try (Admin admin1 = connection1.getAdmin()) {
217      admin1.createTable(table);
218    }
219    try (Admin admin2 = connection2.getAdmin()) {
220      admin2.createTable(table);
221    }
222    utility1.waitUntilAllRegionsAssigned(tablename);
223    utility2.waitUntilAllRegionsAssigned(tablename);
224
225    Table lHtable1 = utility1.getConnection().getTable(tablename);
226
227    // now suspend replication
228    admin.disablePeer("2");
229
230    // put some data (lead with 0 so the edit gets sorted before the other table's edits
231    //   in the replication batch)
232    // write a bunch of edits, making sure we fill a batch
233    byte[] rowkey = Bytes.toBytes(0+" put on table to be dropped");
234    Put put = new Put(rowkey);
235    put.addColumn(familyname, row, row);
236    lHtable1.put(put);
237
238    rowkey = Bytes.toBytes("normal put");
239    put = new Put(rowkey);
240    put.addColumn(famName, row, row);
241    htable1.put(put);
242
243    try (Admin admin2 = connection2.getAdmin()) {
244      admin2.disableTable(tablename);
245      admin2.deleteTable(tablename);
246    }
247
248    admin.enablePeer("2");
249    // edit should still be stuck
250
251    try (Admin admin1 = connection1.getAdmin()) {
252      // the source table still exists, replication should be stalled
253      verifyReplicationStuck(rowkey);
254
255      admin1.disableTable(tablename);
256      // still stuck, source table still exists
257      verifyReplicationStuck(rowkey);
258
259      admin1.deleteTable(tablename);
260      // now the source table is gone, replication should proceed, the
261      // offending edits be dropped
262      verifyReplicationProceeded(rowkey);
263    }
264    // just to be safe
265    conf1.setBoolean(HConstants.REPLICATION_DROP_ON_DELETED_TABLE_KEY, false);
266  }
267
268  private void verifyReplicationProceeded(byte[] rowkey) throws Exception {
269    Get get = new Get(rowkey);
270    for (int i = 0; i < NB_RETRIES; i++) {
271      if (i==NB_RETRIES-1) {
272        fail("Waited too much time for put replication");
273      }
274      Result res = htable2.get(get);
275      if (res.size() == 0) {
276        LOG.info("Row not available");
277        Thread.sleep(SLEEP_TIME);
278      } else {
279        assertArrayEquals(res.getRow(), rowkey);
280        break;
281      }
282    }
283  }
284
285  private void verifyReplicationStuck(byte[] rowkey) throws Exception {
286    Get get = new Get(rowkey);
287    for (int i = 0; i < NB_RETRIES; i++) {
288      Result res = htable2.get(get);
289      if (res.size() >= 1) {
290        fail("Edit should have been stuck behind dropped tables");
291      } else {
292        LOG.info("Row not replicated, let's wait a bit more...");
293        Thread.sleep(SLEEP_TIME);
294      }
295    }
296  }
297}