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.regionserver.regionreplication; 019 020import static org.junit.jupiter.api.Assertions.assertNotNull; 021import static org.junit.jupiter.api.Assertions.assertThrows; 022import static org.junit.jupiter.api.Assertions.assertTrue; 023 024import java.io.IOException; 025import java.util.Collections; 026import org.apache.hadoop.hbase.HBaseTestingUtil; 027import org.apache.hadoop.hbase.ReplicationPeerNotFoundException; 028import org.apache.hadoop.hbase.ServerName; 029import org.apache.hadoop.hbase.SingleProcessHBaseCluster; 030import org.apache.hadoop.hbase.master.HMaster; 031import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure; 032import org.apache.hadoop.hbase.procedure2.Procedure; 033import org.apache.hadoop.hbase.replication.ReplicationGroupOffset; 034import org.apache.hadoop.hbase.replication.ReplicationPeerConfig; 035import org.apache.hadoop.hbase.replication.ReplicationQueueId; 036import org.apache.hadoop.hbase.replication.ReplicationUtils; 037import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceInterface; 038import org.apache.hadoop.hbase.testclassification.MediumTests; 039import org.apache.hadoop.hbase.testclassification.RegionServerTests; 040import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread; 041import org.apache.hadoop.hbase.util.ServerRegionReplicaUtil; 042import org.junit.jupiter.api.AfterAll; 043import org.junit.jupiter.api.BeforeAll; 044import org.junit.jupiter.api.Tag; 045import org.junit.jupiter.api.Test; 046 047/** 048 * Make sure we could start the cluster with RegionReplicaReplicationEndpoint configured. 049 */ 050@Tag(RegionServerTests.TAG) 051@Tag(MediumTests.TAG) 052public class TestStartupWithLegacyRegionReplicationEndpoint { 053 054 private static final HBaseTestingUtil UTIL = new HBaseTestingUtil(); 055 056 @BeforeAll 057 public static void setUp() throws Exception { 058 UTIL.startMiniCluster(1); 059 // add a peer to force initialize the replication storage 060 UTIL.getAdmin().addReplicationPeer("1", ReplicationPeerConfig.newBuilder() 061 .setClusterKey(UTIL.getZkCluster().getAddress().toString() + ":/1").build()); 062 UTIL.getAdmin().removeReplicationPeer("1"); 063 } 064 065 @AfterAll 066 public static void tearDown() throws IOException { 067 UTIL.shutdownMiniCluster(); 068 } 069 070 @Test 071 public void test() throws Exception { 072 String peerId = "legacy"; 073 ReplicationPeerConfig peerConfig = ReplicationPeerConfig.newBuilder() 074 .setClusterKey("127.0.0.1:2181:/hbase") 075 .setReplicationEndpointImpl(ReplicationUtils.LEGACY_REGION_REPLICATION_ENDPOINT_NAME).build(); 076 SingleProcessHBaseCluster cluster = UTIL.getMiniHBaseCluster(); 077 HMaster master = cluster.getMaster(); 078 // can not use Admin.addPeer as it will fail with ClassNotFound 079 master.getReplicationPeerManager().addPeer(peerId, peerConfig, true); 080 // add a wal file to the queue 081 ServerName rsName = cluster.getRegionServer(0).getServerName(); 082 master.getReplicationPeerManager().getQueueStorage().setOffset( 083 new ReplicationQueueId(rsName, ServerRegionReplicaUtil.REGION_REPLICA_REPLICATION_PEER), "", 084 new ReplicationGroupOffset("test-wal-file", 0), Collections.emptyMap()); 085 cluster.stopRegionServer(0); 086 RegionServerThread rst = cluster.startRegionServer(); 087 // we should still have this peer 088 assertNotNull(UTIL.getAdmin().getReplicationPeerConfig(peerId)); 089 // but at RS side, we should not have this peer loaded as replication source 090 assertTrue( 091 rst.getRegionServer().getReplicationSourceService().getReplicationManager().getSources() 092 .stream().map(ReplicationSourceInterface::getPeerId).noneMatch(p -> p.equals(peerId))); 093 094 UTIL.shutdownMiniHBaseCluster(); 095 UTIL.restartHBaseCluster(1); 096 // now we should have removed the peer 097 assertThrows(ReplicationPeerNotFoundException.class, 098 () -> UTIL.getAdmin().getReplicationPeerConfig("legacy")); 099 100 // make sure that we can finish the SCP 101 UTIL.waitFor(15000, 102 () -> UTIL.getMiniHBaseCluster().getMaster().getProcedures().stream() 103 .filter(p -> p instanceof ServerCrashProcedure).map(p -> (ServerCrashProcedure) p) 104 .allMatch(Procedure::isSuccess)); 105 // we will delete the legacy peer while migrating, so here we do not assert the replication data 106 } 107}