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