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.mapred; 019 020import static org.junit.jupiter.api.Assertions.fail; 021 022import java.io.IOException; 023import org.apache.hadoop.hbase.HBaseTestingUtil; 024import org.apache.hadoop.hbase.HConstants; 025import org.apache.hadoop.hbase.io.ImmutableBytesWritable; 026import org.apache.hadoop.hbase.testclassification.MediumTests; 027import org.apache.hadoop.mapred.JobConf; 028import org.apache.hadoop.mapred.RecordWriter; 029import org.junit.jupiter.api.AfterAll; 030import org.junit.jupiter.api.BeforeAll; 031import org.junit.jupiter.api.BeforeEach; 032import org.junit.jupiter.api.Tag; 033import org.junit.jupiter.api.Test; 034import org.slf4j.Logger; 035import org.slf4j.LoggerFactory; 036 037/** 038 * Spark creates many instances of TableOutputFormat within a single process. We need to make sure 039 * we can have many instances and not leak connections. This test creates a few TableOutputFormats 040 * and shouldn't fail due to ZK connection exhaustion. 041 */ 042@Tag(MediumTests.TAG) 043public class TestTableOutputFormatConnectionExhaust { 044 045 private static final Logger LOG = 046 LoggerFactory.getLogger(TestTableOutputFormatConnectionExhaust.class); 047 048 private final static HBaseTestingUtil UTIL = new HBaseTestingUtil(); 049 static final String TABLE = "TestTableOutputFormatConnectionExhaust"; 050 static final String FAMILY = "family"; 051 052 @BeforeAll 053 public static void beforeClass() throws Exception { 054 // Default in ZookeeperMiniCluster is 1000, setting artificially low to trigger exhaustion. 055 // need min of 7 to properly start the default mini HBase cluster 056 UTIL.getConfiguration().setInt(HConstants.ZOOKEEPER_MAX_CLIENT_CNXNS, 10); 057 UTIL.startMiniCluster(); 058 } 059 060 @AfterAll 061 public static void afterClass() throws Exception { 062 UTIL.shutdownMiniCluster(); 063 } 064 065 @BeforeEach 066 public void before() throws IOException { 067 LOG.info("before"); 068 UTIL.ensureSomeRegionServersAvailable(1); 069 LOG.info("before done"); 070 } 071 072 /** 073 * Open and close a TableOutputFormat. The closing the RecordWriter should release HBase 074 * Connection (ZK) resources, and will throw exception if they are exhausted. 075 */ 076 static void openCloseTableOutputFormat(int iter) throws IOException { 077 LOG.info("Instantiating TableOutputFormat connection " + iter); 078 JobConf conf = new JobConf(); 079 conf.addResource(UTIL.getConfiguration()); 080 conf.set(TableOutputFormat.OUTPUT_TABLE, TABLE); 081 TableMapReduceUtil.initTableMapJob(TABLE, FAMILY, TableMap.class, ImmutableBytesWritable.class, 082 ImmutableBytesWritable.class, conf); 083 TableOutputFormat tof = new TableOutputFormat(); 084 RecordWriter rw = tof.getRecordWriter(null, conf, TABLE, null); 085 rw.close(null); 086 } 087 088 @Test 089 public void testConnectionExhaustion() throws IOException { 090 int MAX_INSTANCES = 5; // fails on iteration 3 if zk connections leak 091 for (int i = 0; i < MAX_INSTANCES; i++) { 092 final int iter = i; 093 try { 094 openCloseTableOutputFormat(iter); 095 } catch (Exception e) { 096 LOG.error("Exception encountered", e); 097 fail("Failed on iteration " + i); 098 } 099 } 100 } 101 102}