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.wal; 019 020import java.io.IOException; 021import java.util.List; 022import org.apache.hadoop.conf.Configuration; 023import org.apache.hadoop.fs.FileSystem; 024import org.apache.hadoop.fs.Path; 025import org.apache.hadoop.hbase.HBaseClassTestRule; 026import org.apache.hadoop.hbase.io.asyncfs.monitor.StreamSlowMonitor; 027import org.apache.hadoop.hbase.regionserver.wal.AsyncFSWAL; 028import org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException; 029import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener; 030import org.apache.hadoop.hbase.testclassification.MediumTests; 031import org.apache.hadoop.hbase.testclassification.RegionServerTests; 032import org.apache.hadoop.hbase.util.CommonFSUtils; 033import org.apache.hadoop.hbase.util.Pair; 034import org.junit.AfterClass; 035import org.junit.BeforeClass; 036import org.junit.ClassRule; 037import org.junit.experimental.categories.Category; 038 039import org.apache.hbase.thirdparty.io.netty.channel.Channel; 040import org.apache.hbase.thirdparty.io.netty.channel.EventLoopGroup; 041 042/** 043 * Testcase for HBASE-22539 044 */ 045@Category({ RegionServerTests.class, MediumTests.class }) 046public class TestAsyncFSWALCorruptionDueToDanglingByteBuffer 047 extends WALCorruptionDueToDanglingByteBufferTestBase { 048 049 @ClassRule 050 public static final HBaseClassTestRule CLASS_RULE = 051 HBaseClassTestRule.forClass(TestAsyncFSWALCorruptionDueToDanglingByteBuffer.class); 052 053 public static final class PauseWAL extends AsyncFSWAL { 054 055 public PauseWAL(FileSystem fs, Path rootDir, String logDir, String archiveDir, 056 Configuration conf, List<WALActionsListener> listeners, boolean failIfWALExists, 057 String prefix, String suffix, EventLoopGroup eventLoopGroup, 058 Class<? extends Channel> channelClass) throws FailedLogCloseException, IOException { 059 super(fs, null, rootDir, logDir, archiveDir, conf, listeners, failIfWALExists, prefix, suffix, 060 null, null, eventLoopGroup, channelClass, 061 StreamSlowMonitor.create(conf, "monitorForSuffix")); 062 } 063 064 @Override 065 protected void atHeadOfRingBufferEventHandlerAppend() { 066 if (ARRIVE != null) { 067 ARRIVE.countDown(); 068 try { 069 RESUME.await(); 070 } catch (InterruptedException e) { 071 } 072 } 073 } 074 } 075 076 public static final class PauseWALProvider extends AbstractFSWALProvider<PauseWAL> { 077 078 private EventLoopGroup eventLoopGroup; 079 080 private Class<? extends Channel> channelClass; 081 082 @Override 083 protected PauseWAL createWAL() throws IOException { 084 return new PauseWAL(CommonFSUtils.getWALFileSystem(conf), CommonFSUtils.getWALRootDir(conf), 085 getWALDirectoryName(factory.factoryId), getWALArchiveDirectoryName(conf, factory.factoryId), 086 conf, listeners, true, logPrefix, 087 META_WAL_PROVIDER_ID.equals(providerId) ? META_WAL_PROVIDER_ID : null, eventLoopGroup, 088 channelClass); 089 } 090 091 @Override 092 protected void doInit(Configuration conf) throws IOException { 093 Pair<EventLoopGroup, Class<? extends Channel>> eventLoopGroupAndChannelClass = 094 NettyAsyncFSWALConfigHelper.getEventLoopConfig(conf); 095 eventLoopGroup = eventLoopGroupAndChannelClass.getFirst(); 096 channelClass = eventLoopGroupAndChannelClass.getSecond(); 097 } 098 } 099 100 @BeforeClass 101 public static void setUp() throws Exception { 102 UTIL.getConfiguration().setClass(WALFactory.WAL_PROVIDER, PauseWALProvider.class, 103 WALProvider.class); 104 UTIL.startMiniCluster(1); 105 UTIL.createTable(TABLE_NAME, CF); 106 UTIL.waitTableAvailable(TABLE_NAME); 107 } 108 109 @AfterClass 110 public static void tearDown() throws Exception { 111 UTIL.shutdownMiniCluster(); 112 } 113}