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.mapreduce;
019
020import static org.junit.jupiter.api.Assertions.assertEquals;
021import static org.junit.jupiter.api.Assertions.assertFalse;
022import static org.junit.jupiter.api.Assertions.assertTrue;
023
024import java.io.IOException;
025import java.util.ArrayList;
026import java.util.Collections;
027import java.util.List;
028import org.apache.hadoop.conf.Configuration;
029import org.apache.hadoop.fs.FileStatus;
030import org.apache.hadoop.fs.FileSystem;
031import org.apache.hadoop.fs.LocatedFileStatus;
032import org.apache.hadoop.fs.Path;
033import org.apache.hadoop.hbase.HBaseTestingUtil;
034import org.apache.hadoop.hbase.HConstants;
035import org.apache.hadoop.hbase.regionserver.HRegionServer;
036import org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL;
037import org.apache.hadoop.hbase.testclassification.MapReduceTests;
038import org.apache.hadoop.hbase.testclassification.MediumTests;
039import org.apache.hadoop.hbase.util.CommonFSUtils;
040import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
041import org.apache.hadoop.hdfs.DistributedFileSystem;
042import org.apache.hadoop.mapreduce.InputSplit;
043import org.apache.hadoop.mapreduce.Job;
044import org.apache.hadoop.mapreduce.JobContext;
045import org.apache.hadoop.mapreduce.TaskAttemptContext;
046import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
047import org.junit.jupiter.api.BeforeAll;
048import org.junit.jupiter.api.Tag;
049import org.junit.jupiter.api.Test;
050import org.mockito.Mockito;
051
052@Tag(MapReduceTests.TAG)
053@Tag(MediumTests.TAG)
054public class TestWALInputFormat {
055  private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
056
057  @BeforeAll
058  public static void setupClass() throws Exception {
059    TEST_UTIL.startMiniCluster();
060    TEST_UTIL.createWALRootDir();
061  }
062
063  /**
064   * Test the primitive start/end time filtering.
065   */
066  @Test
067  public void testAddFile() {
068    List<FileStatus> lfss = new ArrayList<>();
069    // a plain FileSystem is never reported closed, so nothing is skipped on the startTime side
070    FileSystem fs = Mockito.mock(FileSystem.class);
071    LocatedFileStatus lfs = Mockito.mock(LocatedFileStatus.class);
072    long now = EnvironmentEdgeManager.currentTime();
073    Mockito.when(lfs.getPath()).thenReturn(new Path("/name." + now));
074    WALInputFormat.addFile(lfss, fs, lfs, now, now);
075    assertEquals(1, lfss.size());
076    WALInputFormat.addFile(lfss, fs, lfs, now - 1, now - 1);
077    assertEquals(1, lfss.size());
078    WALInputFormat.addFile(lfss, fs, lfs, now - 2, now - 1);
079    assertEquals(1, lfss.size());
080    WALInputFormat.addFile(lfss, fs, lfs, now - 2, now);
081    assertEquals(2, lfss.size());
082    WALInputFormat.addFile(lfss, fs, lfs, Long.MIN_VALUE, now);
083    assertEquals(3, lfss.size());
084    WALInputFormat.addFile(lfss, fs, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
085    assertEquals(4, lfss.size());
086    WALInputFormat.addFile(lfss, fs, lfs, now, now + 2);
087    assertEquals(5, lfss.size());
088    // created before startTime, but it may have stayed open and collected in-range entries
089    WALInputFormat.addFile(lfss, fs, lfs, now + 1, now + 2);
090    assertEquals(6, lfss.size());
091    Mockito.when(lfs.getPath()).thenReturn(new Path("/name"));
092    WALInputFormat.addFile(lfss, fs, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
093    assertEquals(7, lfss.size());
094    Mockito.when(lfs.getPath()).thenReturn(new Path("/name.123"));
095    WALInputFormat.addFile(lfss, fs, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
096    assertEquals(8, lfss.size());
097    Mockito.when(lfs.getPath()).thenReturn(new Path("/name." + now + ".meta"));
098    WALInputFormat.addFile(lfss, fs, lfs, now, now);
099    assertEquals(9, lfss.size());
100  }
101
102  private static boolean isKept(FileSystem fs, long created, long mtime, long start, long end) {
103    return isKept(fs, created, mtime, mtime, start, end);
104  }
105
106  private static boolean isKept(FileSystem fs, long created, long staleMtime, long refreshedMtime,
107    long start, long end) {
108    List<FileStatus> result = new ArrayList<>();
109    Path path = new Path("/name." + created);
110    LocatedFileStatus lfs = Mockito.mock(LocatedFileStatus.class);
111    Mockito.when(lfs.getPath()).thenReturn(path);
112    Mockito.when(lfs.getModificationTime()).thenReturn(staleMtime);
113    try {
114      FileStatus refreshed = Mockito.mock(FileStatus.class);
115      Mockito.when(refreshed.getModificationTime()).thenReturn(refreshedMtime);
116      Mockito.when(fs.getFileStatus(path)).thenReturn(refreshed);
117    } catch (IOException e) {
118      throw new RuntimeException(e);
119    }
120    WALInputFormat.addFile(result, fs, lfs, start, end);
121    return !result.isEmpty();
122  }
123
124  /**
125   * The name of a WAL carries its creation time, which only bounds its entries from below. A WAL
126   * stays open until it rolls, so one created before startTime can still hold entries in range.
127   * Only a closed file has a final modification time that can rule it out.
128   */
129  @Test
130  public void testAddFileUsesModificationTimeOfClosedFilesOnly() throws Exception {
131    long now = EnvironmentEdgeManager.currentTime();
132
133    DistributedFileSystem closed = Mockito.mock(DistributedFileSystem.class);
134    Mockito.when(closed.isFileClosed(Mockito.any())).thenReturn(true);
135    DistributedFileSystem open = Mockito.mock(DistributedFileSystem.class);
136    Mockito.when(open.isFileClosed(Mockito.any())).thenReturn(false);
137
138    // Closed, and its last write predates the window: nothing in it can be in range.
139    assertFalse(isKept(closed, now - 100, now - 50, now, now + 100));
140
141    // Same file, but the window opens before it was closed, so it spans the boundary.
142    assertTrue(isKept(closed, now - 100, now - 50, now - 60, now + 100));
143
144    // Still open. Its mtime is stuck near the creation time and says nothing about the entries
145    // it may yet receive, so it has to be kept even though mtime is before the window.
146    assertTrue(isKept(open, now - 100, now - 100, now, now + 100));
147
148    // Created after the window closed: every entry in it is later still.
149    assertFalse(isKept(closed, now + 200, now + 200, now, now + 100));
150
151    // Race condition: file closed between listLocatedStatus and isFileClosed. The stale mtime
152    // from the listing predates the window, but the refreshed mtime (after close) does not.
153    assertTrue(isKept(closed, now - 100, now - 50, now + 10, now, now + 100));
154
155    // ViewDistributedFileSystem wrapping non-HDFS storage: isFileClosed throws
156    // UnsupportedOperationException. The file must be kept (same as non-HDFS).
157    DistributedFileSystem unsupported = Mockito.mock(DistributedFileSystem.class);
158    Mockito.when(unsupported.isFileClosed(Mockito.any()))
159      .thenThrow(new UnsupportedOperationException("mounted fs"));
160    assertTrue(isKept(unsupported, now - 100, now - 50, now, now + 100));
161  }
162
163  @Test
164  public void testHandlesArchivedWALFiles() throws Exception {
165    Configuration conf = TEST_UTIL.getConfiguration();
166    JobContext ctx = Mockito.mock(JobContext.class);
167    Mockito.when(ctx.getConfiguration()).thenReturn(conf);
168    Job job = Job.getInstance(conf);
169    TableMapReduceUtil.initCredentialsForCluster(job, conf);
170    Mockito.when(ctx.getCredentials()).thenReturn(job.getCredentials());
171
172    // Setup WAL file, then archive it
173    HRegionServer rs = TEST_UTIL.getHBaseCluster().getRegionServer(0);
174    AbstractFSWAL wal = (AbstractFSWAL) rs.getWALs().get(0);
175    Path walPath = wal.getCurrentFileName();
176    TEST_UTIL.getConfiguration().set(FileInputFormat.INPUT_DIR, walPath.toString());
177    TEST_UTIL.getConfiguration().set(WALPlayer.INPUT_FILES_SEPARATOR_KEY, ";");
178
179    Path rootDir = CommonFSUtils.getWALRootDir(conf);
180    Path archiveWal = new Path(rootDir, HConstants.HREGION_OLDLOGDIR_NAME);
181    archiveWal = new Path(archiveWal, walPath.getName());
182    TEST_UTIL.getTestFileSystem().delete(walPath, true);
183    TEST_UTIL.getTestFileSystem().mkdirs(archiveWal.getParent());
184    TEST_UTIL.getTestFileSystem().create(archiveWal).close();
185
186    // Test for that we can read from the archived WAL file
187    WALInputFormat wif = new WALInputFormat();
188    List<InputSplit> splits = wif.getSplits(ctx);
189    assertEquals(1, splits.size());
190    WALInputFormat.WALSplit split = (WALInputFormat.WALSplit) splits.get(0);
191    assertEquals(archiveWal.toString(), split.getLogFileName());
192  }
193
194  @Test
195  public void testEmptyFileIsIgnoredWhenConfigured() throws IOException, InterruptedException {
196    List<InputSplit> splits = getSplitsForEmptyFile(true);
197    assertTrue(splits.isEmpty(), "Empty file should be ignored when IGNORE_EMPTY_FILES is true");
198  }
199
200  @Test
201  public void testEmptyFileIsIncludedWhenNotIgnored() throws IOException, InterruptedException {
202    List<InputSplit> splits = getSplitsForEmptyFile(false);
203    assertEquals(1, splits.size(),
204      "Empty file should be included when IGNORE_EMPTY_FILES is false");
205  }
206
207  private List<InputSplit> getSplitsForEmptyFile(boolean ignoreEmptyFiles)
208    throws IOException, InterruptedException {
209    Configuration conf = new Configuration();
210    conf.setBoolean(WALPlayer.IGNORE_EMPTY_FILES, ignoreEmptyFiles);
211
212    JobContext jobContext = Mockito.mock(JobContext.class);
213    Mockito.when(jobContext.getConfiguration()).thenReturn(conf);
214
215    LocatedFileStatus emptyFile = Mockito.mock(LocatedFileStatus.class);
216    Mockito.when(emptyFile.getLen()).thenReturn(0L);
217    Mockito.when(emptyFile.getPath()).thenReturn(new Path("/empty.wal"));
218
219    WALInputFormat inputFormat = new WALInputFormat() {
220      @Override
221      Path[] getInputPaths(Configuration conf) {
222        return new Path[] { new Path("/input") };
223      }
224
225      @Override
226      List<FileStatus> getFiles(FileSystem fs, Path inputPath, long startTime, long endTime,
227        Configuration conf) {
228        return Collections.singletonList(emptyFile);
229      }
230    };
231
232    return inputFormat.getSplits(jobContext, "", "");
233  }
234
235  /**
236   * Test that an empty WAL file (which causes WALHeaderEOFException) is gracefully handled and
237   * skipped rather than causing the job to fail.
238   */
239  @Test
240  public void testHandlesEmptyWALFile() throws Exception {
241    Configuration conf = TEST_UTIL.getConfiguration();
242
243    // Create an empty WAL file
244    Path walRootDir = CommonFSUtils.getWALRootDir(conf);
245    Path emptyWalFile =
246      new Path(walRootDir, "WALs/empty-wal-test/empty." + EnvironmentEdgeManager.currentTime());
247    TEST_UTIL.getTestFileSystem().mkdirs(emptyWalFile.getParent());
248    TEST_UTIL.getTestFileSystem().create(emptyWalFile).close();
249
250    try {
251      JobContext ctx = Mockito.mock(JobContext.class);
252      conf.set(FileInputFormat.INPUT_DIR, emptyWalFile.toString());
253      conf.set(WALPlayer.INPUT_FILES_SEPARATOR_KEY, ";");
254      Mockito.when(ctx.getConfiguration()).thenReturn(conf);
255      Job job = Job.getInstance(conf);
256      TableMapReduceUtil.initCredentialsForCluster(job, conf);
257      Mockito.when(ctx.getCredentials()).thenReturn(job.getCredentials());
258
259      // Create record reader and verify it handles the empty file gracefully
260      try (WALInputFormat.WALKeyRecordReader reader = new WALInputFormat.WALKeyRecordReader()) {
261        TaskAttemptContext taskCtx = Mockito.mock(TaskAttemptContext.class);
262        Mockito.when(taskCtx.getConfiguration()).thenReturn(conf);
263
264        WALInputFormat wif = new WALInputFormat();
265        List<InputSplit> splits = wif.getSplits(ctx);
266        assertEquals(1, splits.size());
267        WALInputFormat.WALSplit split = (WALInputFormat.WALSplit) splits.get(0);
268
269        // This should not throw WALHeaderEOFException - it should return false for nextKeyValue()
270        reader.initialize(split, taskCtx);
271        // nextKeyValue() should return false since the file is empty (reader is null)
272        assertFalse(reader.nextKeyValue());
273      }
274    } finally {
275      TEST_UTIL.getTestFileSystem().delete(emptyWalFile.getParent(), true);
276    }
277  }
278
279}