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; 019 020import static org.junit.jupiter.api.Assertions.assertArrayEquals; 021import static org.junit.jupiter.api.Assertions.assertEquals; 022 023import java.util.Arrays; 024import java.util.List; 025import java.util.stream.Collectors; 026import java.util.stream.IntStream; 027import org.apache.hadoop.hbase.HBaseTestingUtil; 028import org.apache.hadoop.hbase.HConstants; 029import org.apache.hadoop.hbase.TableName; 030import org.apache.hadoop.hbase.logging.Log4jUtils; 031import org.apache.hadoop.hbase.testclassification.ClientTests; 032import org.apache.hadoop.hbase.testclassification.MediumTests; 033import org.apache.hadoop.hbase.util.Bytes; 034import org.junit.jupiter.api.AfterAll; 035import org.junit.jupiter.api.BeforeAll; 036import org.junit.jupiter.api.Tag; 037import org.junit.jupiter.api.Test; 038 039@Tag(MediumTests.TAG) 040@Tag(ClientTests.TAG) 041public class TestAsyncTableBatchRetryImmediately { 042 043 private static final HBaseTestingUtil UTIL = new HBaseTestingUtil(); 044 045 private static TableName TABLE_NAME = TableName.valueOf("async"); 046 047 private static byte[] FAMILY = Bytes.toBytes("cf"); 048 049 private static byte[] QUAL = Bytes.toBytes("cq"); 050 051 private static byte[] VALUE_PREFIX = new byte[768]; 052 053 private static int COUNT = 1000; 054 055 private static AsyncConnection CONN; 056 057 private static String LOG_LEVEL; 058 059 @BeforeAll 060 public static void setUp() throws Exception { 061 // disable the debug log to avoid flooding the output 062 LOG_LEVEL = Log4jUtils.getEffectiveLevel(AsyncRegionLocatorHelper.class.getName()); 063 Log4jUtils.setLogLevel(AsyncRegionLocatorHelper.class.getName(), "INFO"); 064 UTIL.getConfiguration().setLong(HConstants.HBASE_SERVER_SCANNER_MAX_RESULT_SIZE_KEY, 1024); 065 UTIL.startMiniCluster(1); 066 Table table = UTIL.createTable(TABLE_NAME, FAMILY); 067 UTIL.waitTableAvailable(TABLE_NAME); 068 Bytes.random(VALUE_PREFIX); 069 for (int i = 0; i < COUNT; i++) { 070 table.put(new Put(Bytes.toBytes(i)).addColumn(FAMILY, QUAL, 071 Bytes.add(VALUE_PREFIX, Bytes.toBytes(i)))); 072 } 073 CONN = ConnectionFactory.createAsyncConnection(UTIL.getConfiguration()).get(); 074 } 075 076 @AfterAll 077 public static void tearDown() throws Exception { 078 if (LOG_LEVEL != null) { 079 Log4jUtils.setLogLevel(AsyncRegionLocatorHelper.class.getName(), LOG_LEVEL); 080 } 081 CONN.close(); 082 UTIL.shutdownMiniCluster(); 083 } 084 085 @Test 086 public void test() { 087 AsyncTable<?> table = CONN.getTable(TABLE_NAME); 088 // if we do not deal with RetryImmediatelyException, we will timeout here since we need to retry 089 // hundreds times. 090 List<Get> gets = IntStream.range(0, COUNT).mapToObj(i -> new Get(Bytes.toBytes(i))) 091 .collect(Collectors.toList()); 092 List<Result> results = table.getAll(gets).join(); 093 for (int i = 0; i < COUNT; i++) { 094 byte[] value = results.get(i).getValue(FAMILY, QUAL); 095 assertEquals(VALUE_PREFIX.length + 4, value.length); 096 assertArrayEquals(VALUE_PREFIX, Arrays.copyOf(value, VALUE_PREFIX.length)); 097 assertEquals(i, Bytes.toInt(value, VALUE_PREFIX.length)); 098 } 099 } 100}