hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestFallbackToUseReplay.java

   1 /**
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.client;
  19
  20 import static org.junit.Assert.assertTrue;
  21 import static org.mockito.ArgumentMatchers.any;
  22 import static org.mockito.ArgumentMatchers.anyInt;
  23 import static org.mockito.ArgumentMatchers.anyLong;
  24 import static org.mockito.Mockito.doAnswer;
  25 import static org.mockito.Mockito.mock;
  26 import static org.mockito.Mockito.when;
  27
  28 import java.io.IOException;
  29 import java.util.Collections;
  30 import java.util.concurrent.CompletableFuture;
  31 import java.util.concurrent.TimeUnit;
  32 import java.util.concurrent.atomic.AtomicBoolean;
  33 import org.apache.hadoop.conf.Configuration;
  34 import org.apache.hadoop.hbase.DoNotRetryIOException;
  35 import org.apache.hadoop.hbase.HBaseClassTestRule;
  36 import org.apache.hadoop.hbase.HBaseConfiguration;
  37 import org.apache.hadoop.hbase.HRegionLocation;
  38 import org.apache.hadoop.hbase.ServerName;
  39 import org.apache.hadoop.hbase.TableName;
  40 import org.apache.hadoop.hbase.ipc.HBaseRpcController;
  41 import org.apache.hadoop.hbase.security.User;
  42 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
  43 import org.apache.hadoop.hbase.testclassification.SmallTests;
  44 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
  45 import org.junit.AfterClass;
  46 import org.junit.BeforeClass;
  47 import org.junit.ClassRule;
  48 import org.junit.Test;
  49 import org.junit.experimental.categories.Category;
  50
  51 import org.apache.hbase.thirdparty.com.google.common.io.Closeables;
  52 import org.apache.hbase.thirdparty.com.google.protobuf.RpcCallback;
  53
  54 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
  55 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService.Interface;
  56
  57 /**
  58  * Make sure we could fallback to use replay method if replicateToReplica method is not present,
  59  * i.e, we are connecting an old region server.
  60  */
  61 @Category({ RegionServerTests.class, SmallTests.class })
  62 public class TestFallbackToUseReplay {
  63
  64   @ClassRule
  65   public static final HBaseClassTestRule CLASS_RULE =
  66     HBaseClassTestRule.forClass(TestFallbackToUseReplay.class);
  67
  68   private static Configuration CONF = HBaseConfiguration.create();
  69
  70   private static AsyncClusterConnectionImpl CONN;
  71
  72   private static AsyncRegionReplicationRetryingCaller CALLER;
  73
  74   private static RegionInfo REPLICA =
  75     RegionInfoBuilder.newBuilder(TableName.valueOf("test")).setReplicaId(1).build();
  76
  77   private static AtomicBoolean REPLAY_CALLED = new AtomicBoolean(false);
  78
  79   @BeforeClass
  80   public static void setUpBeforeClass() throws IOException {
  81     CONF.setInt(AsyncConnectionConfiguration.START_LOG_ERRORS_AFTER_COUNT_KEY, 0);
  82     AsyncRegionLocator locator = mock(AsyncRegionLocator.class);
  83     when(locator.getRegionLocation(any(), any(), anyInt(), any(), anyLong()))
  84       .thenReturn(CompletableFuture.completedFuture(new HRegionLocation(REPLICA,
  85         ServerName.valueOf("localhost", 12345, EnvironmentEdgeManager.currentTime()))));
  86     AdminService.Interface stub = mock(AdminService.Interface.class);
  87     // fail the call to replicateToReplica
  88     doAnswer(i -> {
  89       HBaseRpcController controller = i.getArgument(0, HBaseRpcController.class);
  90       controller.setFailed(new DoNotRetryIOException(new UnsupportedOperationException()));
  91       RpcCallback<?> done = i.getArgument(2, RpcCallback.class);
  92       done.run(null);
  93       return null;
  94     }).when(stub).replicateToReplica(any(), any(), any());
  95     doAnswer(i -> {
  96       REPLAY_CALLED.set(true);
  97       RpcCallback<?> done = i.getArgument(2, RpcCallback.class);
  98       done.run(null);
  99       return null;
 100     }).when(stub).replay(any(), any(), any());
 101     CONN = new AsyncClusterConnectionImpl(CONF, mock(ConnectionRegistry.class), "test", null,
 102       User.getCurrent()) {
 103
 104       @Override
 105       AsyncRegionLocator getLocator() {
 106         return locator;
 107       }
 108
 109       @Override
 110       Interface getAdminStub(ServerName serverName) throws IOException {
 111         return stub;
 112       }
 113     };
 114     CALLER = new AsyncRegionReplicationRetryingCaller(AsyncClusterConnectionImpl.RETRY_TIMER, CONN,
 115       10, TimeUnit.SECONDS.toNanos(1), TimeUnit.SECONDS.toNanos(10), REPLICA,
 116       Collections.emptyList());
 117   }
 118
 119   @AfterClass
 120   public static void tearDownAfterClass() throws IOException {
 121     Closeables.close(CONN, true);
 122   }
 123
 124   @Test
 125   public void testFallback() {
 126     CALLER.call().join();
 127     assertTrue(REPLAY_CALLED.get());
 128   }
 129 }