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.client.replication;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertFalse;
022import static org.junit.Assert.assertTrue;
023import static org.junit.Assert.fail;
024
025import java.util.Collection;
026import java.util.HashMap;
027import java.util.Map;
028import java.util.UUID;
029import org.apache.hadoop.hbase.HBaseClassTestRule;
030import org.apache.hadoop.hbase.HColumnDescriptor;
031import org.apache.hadoop.hbase.HConstants;
032import org.apache.hadoop.hbase.HTableDescriptor;
033import org.apache.hadoop.hbase.TableName;
034import org.apache.hadoop.hbase.TableNotFoundException;
035import org.apache.hadoop.hbase.client.Admin;
036import org.apache.hadoop.hbase.client.Connection;
037import org.apache.hadoop.hbase.client.ConnectionFactory;
038import org.apache.hadoop.hbase.replication.BaseReplicationEndpoint;
039import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
040import org.apache.hadoop.hbase.replication.TestReplicationBase;
041import org.apache.hadoop.hbase.testclassification.ClientTests;
042import org.apache.hadoop.hbase.testclassification.MediumTests;
043import org.junit.AfterClass;
044import org.junit.BeforeClass;
045import org.junit.ClassRule;
046import org.junit.Rule;
047import org.junit.Test;
048import org.junit.experimental.categories.Category;
049import org.junit.rules.TestName;
050
051/**
052 * Unit testing of ReplicationAdmin with clusters
053 */
054@Category({ MediumTests.class, ClientTests.class })
055public class TestReplicationAdminWithClusters extends TestReplicationBase {
056
057  @ClassRule
058  public static final HBaseClassTestRule CLASS_RULE =
059      HBaseClassTestRule.forClass(TestReplicationAdminWithClusters.class);
060
061  static Connection connection1;
062  static Connection connection2;
063  static Admin admin1;
064  static Admin admin2;
065  static ReplicationAdmin adminExt;
066
067  @Rule
068  public TestName name = new TestName();
069
070  @BeforeClass
071  public static void setUpBeforeClass() throws Exception {
072    TestReplicationBase.setUpBeforeClass();
073    connection1 = ConnectionFactory.createConnection(conf1);
074    connection2 = ConnectionFactory.createConnection(conf2);
075    admin1 = connection1.getAdmin();
076    admin2 = connection2.getAdmin();
077    adminExt = new ReplicationAdmin(conf1);
078  }
079
080  @AfterClass
081  public static void tearDownAfterClass() throws Exception {
082    admin1.close();
083    admin2.close();
084    adminExt.close();
085    connection1.close();
086    connection2.close();
087    TestReplicationBase.tearDownAfterClass();
088  }
089
090  @Test
091  public void disableNotFullReplication() throws Exception {
092    HTableDescriptor table = new HTableDescriptor(admin2.getTableDescriptor(tableName));
093    HColumnDescriptor f = new HColumnDescriptor("notReplicatedFamily");
094    table.addFamily(f);
095    admin1.disableTable(tableName);
096    admin1.modifyTable(tableName, table);
097    admin1.enableTable(tableName);
098
099
100    admin1.disableTableReplication(tableName);
101    table = admin1.getTableDescriptor(tableName);
102    for (HColumnDescriptor fam : table.getColumnFamilies()) {
103      assertEquals(HConstants.REPLICATION_SCOPE_LOCAL, fam.getScope());
104    }
105  }
106
107  @Test
108  public void testEnableReplicationWhenSlaveClusterDoesntHaveTable() throws Exception {
109    admin1.disableTableReplication(tableName);
110    admin2.disableTable(tableName);
111    admin2.deleteTable(tableName);
112    assertFalse(admin2.tableExists(tableName));
113    admin1.enableTableReplication(tableName);
114    assertTrue(admin2.tableExists(tableName));
115  }
116
117  @Test
118  public void testEnableReplicationWhenReplicationNotEnabled() throws Exception {
119    HTableDescriptor table = new HTableDescriptor(admin1.getTableDescriptor(tableName));
120    for (HColumnDescriptor fam : table.getColumnFamilies()) {
121      fam.setScope(HConstants.REPLICATION_SCOPE_LOCAL);
122    }
123    admin1.disableTable(tableName);
124    admin1.modifyTable(tableName, table);
125    admin1.enableTable(tableName);
126
127    admin2.disableTable(tableName);
128    admin2.modifyTable(tableName, table);
129    admin2.enableTable(tableName);
130
131    admin1.enableTableReplication(tableName);
132    table = admin1.getTableDescriptor(tableName);
133    for (HColumnDescriptor fam : table.getColumnFamilies()) {
134      assertEquals(HConstants.REPLICATION_SCOPE_GLOBAL, fam.getScope());
135    }
136  }
137
138  @Test
139  public void testEnableReplicationWhenTableDescriptorIsNotSameInClusters() throws Exception {
140    HTableDescriptor table = new HTableDescriptor(admin2.getTableDescriptor(tableName));
141    HColumnDescriptor f = new HColumnDescriptor("newFamily");
142    table.addFamily(f);
143    admin2.disableTable(tableName);
144    admin2.modifyTable(tableName, table);
145    admin2.enableTable(tableName);
146
147    try {
148      admin1.enableTableReplication(tableName);
149      fail("Exception should be thrown if table descriptors in the clusters are not same.");
150    } catch (RuntimeException ignored) {
151
152    }
153    admin1.disableTable(tableName);
154    admin1.modifyTable(tableName, table);
155    admin1.enableTable(tableName);
156    admin1.enableTableReplication(tableName);
157    table = admin1.getTableDescriptor(tableName);
158    for (HColumnDescriptor fam : table.getColumnFamilies()) {
159      assertEquals(HConstants.REPLICATION_SCOPE_GLOBAL, fam.getScope());
160    }
161  }
162
163  @Test
164  public void testDisableAndEnableReplication() throws Exception {
165    admin1.disableTableReplication(tableName);
166    HTableDescriptor table = admin1.getTableDescriptor(tableName);
167    for (HColumnDescriptor fam : table.getColumnFamilies()) {
168      assertEquals(HConstants.REPLICATION_SCOPE_LOCAL, fam.getScope());
169    }
170    admin1.enableTableReplication(tableName);
171    table = admin1.getTableDescriptor(tableName);
172    for (HColumnDescriptor fam : table.getColumnFamilies()) {
173      assertEquals(HConstants.REPLICATION_SCOPE_GLOBAL, fam.getScope());
174    }
175  }
176
177  @Test(expected = TableNotFoundException.class)
178  public void testDisableReplicationForNonExistingTable() throws Exception {
179    admin1.disableTableReplication(TableName.valueOf(name.getMethodName()));
180  }
181
182  @Test(expected = TableNotFoundException.class)
183  public void testEnableReplicationForNonExistingTable() throws Exception {
184    admin1.enableTableReplication(TableName.valueOf(name.getMethodName()));
185  }
186
187  @Test(expected = IllegalArgumentException.class)
188  public void testDisableReplicationWhenTableNameAsNull() throws Exception {
189    admin1.disableTableReplication(null);
190  }
191
192  @Test(expected = IllegalArgumentException.class)
193  public void testEnableReplicationWhenTableNameAsNull() throws Exception {
194    admin1.enableTableReplication(null);
195  }
196
197  /*
198   * Test enable table replication should create table only in user explicit specified table-cfs.
199   * HBASE-14717
200   */
201  @Test
202  public void testEnableReplicationForExplicitSetTableCfs() throws Exception {
203    final TableName tableName = TableName.valueOf(name.getMethodName());
204    String peerId = "2";
205    if (admin2.isTableAvailable(TestReplicationBase.tableName)) {
206      admin2.disableTable(TestReplicationBase.tableName);
207      admin2.deleteTable(TestReplicationBase.tableName);
208    }
209    assertFalse("Table should not exists in the peer cluster",
210      admin2.isTableAvailable(TestReplicationBase.tableName));
211
212    // update peer config
213    ReplicationPeerConfig rpc = admin1.getReplicationPeerConfig(peerId);
214    rpc.setReplicateAllUserTables(false);
215    admin1.updateReplicationPeerConfig(peerId, rpc);
216
217    Map<TableName, ? extends Collection<String>> tableCfs = new HashMap<>();
218    tableCfs.put(tableName, null);
219    try {
220      adminExt.setPeerTableCFs(peerId, tableCfs);
221      admin1.enableTableReplication(TestReplicationBase.tableName);
222      assertFalse("Table should not be created if user has set table cfs explicitly for the "
223          + "peer and this is not part of that collection",
224        admin2.isTableAvailable(TestReplicationBase.tableName));
225
226      tableCfs.put(TestReplicationBase.tableName, null);
227      adminExt.setPeerTableCFs(peerId, tableCfs);
228      admin1.enableTableReplication(TestReplicationBase.tableName);
229      assertTrue(
230        "Table should be created if user has explicitly added table into table cfs collection",
231        admin2.isTableAvailable(TestReplicationBase.tableName));
232    } finally {
233      adminExt.removePeerTableCFs(peerId, adminExt.getPeerTableCFs(peerId));
234      admin1.disableTableReplication(TestReplicationBase.tableName);
235
236      rpc = admin1.getReplicationPeerConfig(peerId);
237      rpc.setReplicateAllUserTables(true);
238      admin1.updateReplicationPeerConfig(peerId, rpc);
239    }
240  }
241
242  @Test
243  public void testReplicationPeerConfigUpdateCallback() throws Exception {
244    String peerId = "1";
245    ReplicationPeerConfig rpc = new ReplicationPeerConfig();
246    rpc.setClusterKey(utility2.getClusterKey());
247    rpc.setReplicationEndpointImpl(TestUpdatableReplicationEndpoint.class.getName());
248    rpc.getConfiguration().put("key1", "value1");
249
250    admin1.addReplicationPeer(peerId, rpc);
251
252    rpc.getConfiguration().put("key1", "value2");
253    admin.updatePeerConfig(peerId, rpc);
254    if (!TestUpdatableReplicationEndpoint.hasCalledBack()) {
255      synchronized(TestUpdatableReplicationEndpoint.class) {
256        TestUpdatableReplicationEndpoint.class.wait(2000L);
257      }
258    }
259
260    assertEquals(true, TestUpdatableReplicationEndpoint.hasCalledBack());
261  }
262
263  public static class TestUpdatableReplicationEndpoint extends BaseReplicationEndpoint {
264    private static boolean calledBack = false;
265    public static boolean hasCalledBack(){
266      return calledBack;
267    }
268    @Override
269    public synchronized void peerConfigUpdated(ReplicationPeerConfig rpc){
270      calledBack = true;
271      notifyAll();
272    }
273
274    @Override
275    public void start() {
276      startAsync();
277    }
278
279    @Override
280    public void stop() {
281      stopAsync();
282    }
283
284    @Override
285    protected void doStart() {
286      notifyStarted();
287    }
288
289    @Override
290    protected void doStop() {
291      notifyStopped();
292    }
293
294
295    @Override
296    public UUID getPeerUUID() {
297      return UUID.randomUUID();
298    }
299
300    @Override
301    public boolean replicate(ReplicateContext replicateContext) {
302      return false;
303    }
304  }
305}