Source code

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.wal;
019
020import static org.junit.jupiter.api.Assertions.assertEquals;
021import static org.junit.jupiter.api.Assertions.assertNotNull;
022import static org.mockito.Mockito.mock;
023import static org.mockito.Mockito.when;
024
025import java.io.IOException;
026import java.lang.reflect.Field;
027import java.util.List;
028import java.util.NavigableMap;
029import java.util.TreeMap;
030import java.util.concurrent.CountDownLatch;
031import java.util.concurrent.ExecutorService;
032import java.util.concurrent.Executors;
033import java.util.concurrent.atomic.AtomicBoolean;
034import org.apache.hadoop.conf.Configuration;
035import org.apache.hadoop.fs.FSDataOutputStream;
036import org.apache.hadoop.fs.FileSystem;
037import org.apache.hadoop.fs.Path;
038import org.apache.hadoop.hbase.HConstants;
039import org.apache.hadoop.hbase.TableName;
040import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
041import org.apache.hadoop.hbase.client.Put;
042import org.apache.hadoop.hbase.client.RegionInfo;
043import org.apache.hadoop.hbase.client.RegionInfoBuilder;
044import org.apache.hadoop.hbase.client.TableDescriptor;
045import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
046import org.apache.hadoop.hbase.regionserver.ChunkCreator;
047import org.apache.hadoop.hbase.regionserver.HRegion;
048import org.apache.hadoop.hbase.regionserver.MemStoreLAB;
049import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
050import org.apache.hadoop.hbase.testclassification.MediumTests;
051import org.apache.hadoop.hbase.testclassification.RegionServerTests;
052import org.apache.hadoop.hbase.util.Bytes;
053import org.apache.hadoop.hbase.util.CommonFSUtils;
054import org.apache.hadoop.hbase.util.Threads;
055import org.apache.hadoop.hbase.wal.WALEdit;
056import org.apache.hadoop.hbase.wal.WALKey;
057import org.apache.hadoop.hdfs.DFSOutputStream;
058import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
059import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
060import org.junit.jupiter.api.BeforeEach;
061import org.junit.jupiter.api.Tag;
062import org.junit.jupiter.api.Test;
063import org.junit.jupiter.api.TestInfo;
064
065/**
066 * Provides FSHLog test cases.
067 */
068@Tag(RegionServerTests.TAG)
069@Tag(MediumTests.TAG)
070public class TestFSHLog extends AbstractTestFSWAL {
071
072  private String name;
073
074  @BeforeEach
075  public void initTestName(TestInfo testInfo) {
076    name = testInfo.getTestMethod().get().getName();
077  }
078
079  @Override
080  protected AbstractFSWAL<?> newWAL(FileSystem fs, Path rootDir, String walDir, String archiveDir,
081    Configuration conf, List<WALActionsListener> listeners, boolean failIfWALExists, String prefix,
082    String suffix) throws IOException {
083    FSHLog wal =
084      new FSHLog(fs, rootDir, walDir, archiveDir, conf, listeners, failIfWALExists, prefix, suffix);
085    wal.init();
086    return wal;
087  }
088
089  @Override
090  protected AbstractFSWAL<?> newSlowWAL(FileSystem fs, Path rootDir, String walDir,
091    String archiveDir, Configuration conf, List<WALActionsListener> listeners,
092    boolean failIfWALExists, String prefix, String suffix, final Runnable action)
093    throws IOException {
094    FSHLog wal = new FSHLog(fs, rootDir, walDir, archiveDir, conf, listeners, failIfWALExists,
095      prefix, suffix) {
096
097      @Override
098      protected void atHeadOfRingBufferEventHandlerAppend() {
099        action.run();
100        super.atHeadOfRingBufferEventHandlerAppend();
101      }
102    };
103    wal.init();
104    return wal;
105  }
106
107  @Test
108  public void testSyncRunnerIndexOverflow() throws IOException, NoSuchFieldException,
109    SecurityException, IllegalArgumentException, IllegalAccessException {
110    FS.mkdirs(new Path(CommonFSUtils.getRootDir(CONF), this.name));
111    FSHLog log = new FSHLog(FS, CommonFSUtils.getRootDir(CONF), this.name,
112      HConstants.HREGION_OLDLOGDIR_NAME, CONF, null, true, null, null);
113    log.init();
114    try {
115      Field syncRunnerIndexField = FSHLog.class.getDeclaredField("syncRunnerIndex");
116      syncRunnerIndexField.setAccessible(true);
117      syncRunnerIndexField.set(log, Integer.MAX_VALUE - 1);
118      TableDescriptor htd = TableDescriptorBuilder.newBuilder(TableName.valueOf(this.name))
119        .setColumnFamily(ColumnFamilyDescriptorBuilder.of("row")).build();
120      NavigableMap<byte[], Integer> scopes = new TreeMap<>(Bytes.BYTES_COMPARATOR);
121      for (byte[] fam : htd.getColumnFamilyNames()) {
122        scopes.put(fam, 0);
123      }
124      RegionInfo hri = RegionInfoBuilder.newBuilder(htd.getTableName()).build();
125      MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
126      for (int i = 0; i < 10; i++) {
127        addEdits(log, hri, htd, 1, mvcc, scopes, "row");
128      }
129    } finally {
130      log.close();
131    }
132  }
133
134  /**
135   * Test case for https://issues.apache.org/jira/browse/HBASE-16721
136   */
137  @Test
138  public void testUnflushedSeqIdTracking() throws IOException, InterruptedException {
139    final byte[] b = Bytes.toBytes("b");
140
141    final AtomicBoolean startHoldingForAppend = new AtomicBoolean(false);
142    final CountDownLatch holdAppend = new CountDownLatch(1);
143    final CountDownLatch flushFinished = new CountDownLatch(1);
144    final CountDownLatch putFinished = new CountDownLatch(1);
145
146    FS.mkdirs(new Path(CommonFSUtils.getRootDir(CONF), this.name));
147    try (FSHLog log = new FSHLog(FS, CommonFSUtils.getRootDir(CONF), this.name,
148      HConstants.HREGION_OLDLOGDIR_NAME, CONF, null, true, null, null)) {
149      log.init();
150      log.registerWALActionsListener(new WALActionsListener() {
151        @Override
152        public void visitLogEntryBeforeWrite(RegionInfo info, WALKey logKey, WALEdit logEdit) {
153          if (startHoldingForAppend.get()) {
154            try {
155              holdAppend.await();
156            } catch (InterruptedException e) {
157              LOG.error(e.toString(), e);
158            }
159          }
160        }
161      });
162
163      // open a new region which uses this WAL
164      TableDescriptor htd = TableDescriptorBuilder.newBuilder(TableName.valueOf(this.name))
165        .setColumnFamily(ColumnFamilyDescriptorBuilder.of(b)).build();
166      RegionInfo hri = RegionInfoBuilder.newBuilder(htd.getTableName()).build();
167      ChunkCreator.initialize(MemStoreLAB.CHUNK_SIZE_DEFAULT, false, 0, 0, 0, null,
168        MemStoreLAB.INDEX_CHUNK_SIZE_PERCENTAGE_DEFAULT);
169      final HRegion region = TEST_UTIL.createLocalHRegion(hri, CONF, htd, log);
170      ExecutorService exec = Executors.newFixedThreadPool(2);
171
172      // do a regular write first because of memstore size calculation.
173      region.put(new Put(b).addColumn(b, b, b));
174
175      startHoldingForAppend.set(true);
176      exec.submit(new Runnable() {
177        @Override
178        public void run() {
179          try {
180            region.put(new Put(b).addColumn(b, b, b));
181            putFinished.countDown();
182          } catch (IOException e) {
183            LOG.error(e.toString(), e);
184          }
185        }
186      });
187
188      // give the put a chance to start
189      Threads.sleep(3000);
190
191      exec.submit(new Runnable() {
192        @Override
193        public void run() {
194          try {
195            HRegion.FlushResult flushResult = region.flush(true);
196            LOG.info("Flush result:" + flushResult.getResult());
197            LOG.info("Flush succeeded:" + flushResult.isFlushSucceeded());
198            flushFinished.countDown();
199          } catch (IOException e) {
200            LOG.error(e.toString(), e);
201          }
202        }
203      });
204
205      // give the flush a chance to start. Flush should have got the region lock, and
206      // should have been waiting on the mvcc complete after this.
207      Threads.sleep(3000);
208
209      // let the append to WAL go through now that the flush already started
210      holdAppend.countDown();
211      putFinished.await();
212      flushFinished.await();
213
214      // check whether flush went through
215      assertEquals(1, region.getStoreFileList(new byte[][] { b }).size(), "Region did not flush?");
216
217      // now check the region's unflushed seqIds.
218      long seqId = AbstractTestFSWAL.getEarliestMemStoreSeqNum(log, hri.getEncodedNameAsBytes());
219      assertEquals(HConstants.NO_SEQNUM, seqId,
220        "Found seqId for the region which is already flushed");
221
222      region.close();
223    }
224  }
225
226  /**
227   * Regression test for HBASE-30346: FSHLog#getPipeline() must never return null, even when the
228   * underlying DFSOutputStream#getPipeline() legitimately returns null (e.g. the DFS streamer is
229   * closed, or no block pipeline is currently established -- see HDFS-826 and the
230   * DFSOutputStream#getPipeline() javadoc: "returns the list of targets, if any"). Prior to this
231   * fix, AbstractFSWAL#rollWriterInternal's debug-log statement called
232   * Arrays.stream(getPipeline()), which threw a NullPointerException whenever this happened during
233   * a WAL roll, aborting the RegionServer on what was otherwise a successful roll.
234   */
235  @Test
236  public void testGetPipelineDoesNotReturnNullWhenUnderlyingStreamerHasNone() throws Exception {
237    FS.mkdirs(new Path(CommonFSUtils.getRootDir(CONF), this.name));
238    try (FSHLog log = new FSHLog(FS, CommonFSUtils.getRootDir(CONF), this.name,
239      HConstants.HREGION_OLDLOGDIR_NAME, CONF, null, true, null, null)) {
240      log.init();
241
242      // Simulate the legitimate HDFS contract: the wrapped DFSOutputStream currently has no
243      // established pipeline (e.g. streamer closed, or between blocks) and returns null.
244      DFSOutputStream mockDfsOut = mock(DFSOutputStream.class);
245      when(mockDfsOut.getPipeline()).thenReturn(null);
246      FSDataOutputStream wrappedOut = new HdfsDataOutputStream(mockDfsOut, null);
247
248      Field hdfsOutField = FSHLog.class.getDeclaredField("hdfs_out");
249      hdfsOutField.setAccessible(true);
250      hdfsOutField.set(log, wrappedOut);
251
252      DatanodeInfo[] pipeline = log.getPipeline();
253      assertNotNull(pipeline, "getPipeline() must never return null");
254      assertEquals(0, pipeline.length,
255        "Should normalize a null underlying pipeline to an empty array");
256    }
257  }
258}