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.mapreduce.InputSplit;
042import org.apache.hadoop.mapreduce.Job;
043import org.apache.hadoop.mapreduce.JobContext;
044import org.apache.hadoop.mapreduce.TaskAttemptContext;
045import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
046import org.junit.jupiter.api.BeforeAll;
047import org.junit.jupiter.api.Tag;
048import org.junit.jupiter.api.Test;
049import org.mockito.Mockito;
050
051@Tag(MapReduceTests.TAG)
052@Tag(MediumTests.TAG)
053public class TestWALInputFormat {
054  private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
055
056  @BeforeAll
057  public static void setupClass() throws Exception {
058    TEST_UTIL.startMiniCluster();
059    TEST_UTIL.createWALRootDir();
060  }
061
062  /**
063   * Test the primitive start/end time filtering.
064   */
065  @Test
066  public void testAddFile() {
067    List<FileStatus> lfss = new ArrayList<>();
068    LocatedFileStatus lfs = Mockito.mock(LocatedFileStatus.class);
069    long now = EnvironmentEdgeManager.currentTime();
070    Mockito.when(lfs.getPath()).thenReturn(new Path("/name." + now));
071    WALInputFormat.addFile(lfss, lfs, now, now);
072    assertEquals(1, lfss.size());
073    WALInputFormat.addFile(lfss, lfs, now - 1, now - 1);
074    assertEquals(1, lfss.size());
075    WALInputFormat.addFile(lfss, lfs, now - 2, now - 1);
076    assertEquals(1, lfss.size());
077    WALInputFormat.addFile(lfss, lfs, now - 2, now);
078    assertEquals(2, lfss.size());
079    WALInputFormat.addFile(lfss, lfs, Long.MIN_VALUE, now);
080    assertEquals(3, lfss.size());
081    WALInputFormat.addFile(lfss, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
082    assertEquals(4, lfss.size());
083    WALInputFormat.addFile(lfss, lfs, now, now + 2);
084    assertEquals(5, lfss.size());
085    WALInputFormat.addFile(lfss, lfs, now + 1, now + 2);
086    assertEquals(5, lfss.size());
087    Mockito.when(lfs.getPath()).thenReturn(new Path("/name"));
088    WALInputFormat.addFile(lfss, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
089    assertEquals(6, lfss.size());
090    Mockito.when(lfs.getPath()).thenReturn(new Path("/name.123"));
091    WALInputFormat.addFile(lfss, lfs, Long.MIN_VALUE, Long.MAX_VALUE);
092    assertEquals(7, lfss.size());
093    Mockito.when(lfs.getPath()).thenReturn(new Path("/name." + now + ".meta"));
094    WALInputFormat.addFile(lfss, lfs, now, now);
095    assertEquals(8, lfss.size());
096  }
097
098  @Test
099  public void testHandlesArchivedWALFiles() throws Exception {
100    Configuration conf = TEST_UTIL.getConfiguration();
101    JobContext ctx = Mockito.mock(JobContext.class);
102    Mockito.when(ctx.getConfiguration()).thenReturn(conf);
103    Job job = Job.getInstance(conf);
104    TableMapReduceUtil.initCredentialsForCluster(job, conf);
105    Mockito.when(ctx.getCredentials()).thenReturn(job.getCredentials());
106
107    // Setup WAL file, then archive it
108    HRegionServer rs = TEST_UTIL.getHBaseCluster().getRegionServer(0);
109    AbstractFSWAL wal = (AbstractFSWAL) rs.getWALs().get(0);
110    Path walPath = wal.getCurrentFileName();
111    TEST_UTIL.getConfiguration().set(FileInputFormat.INPUT_DIR, walPath.toString());
112    TEST_UTIL.getConfiguration().set(WALPlayer.INPUT_FILES_SEPARATOR_KEY, ";");
113
114    Path rootDir = CommonFSUtils.getWALRootDir(conf);
115    Path archiveWal = new Path(rootDir, HConstants.HREGION_OLDLOGDIR_NAME);
116    archiveWal = new Path(archiveWal, walPath.getName());
117    TEST_UTIL.getTestFileSystem().delete(walPath, true);
118    TEST_UTIL.getTestFileSystem().mkdirs(archiveWal.getParent());
119    TEST_UTIL.getTestFileSystem().create(archiveWal).close();
120
121    // Test for that we can read from the archived WAL file
122    WALInputFormat wif = new WALInputFormat();
123    List<InputSplit> splits = wif.getSplits(ctx);
124    assertEquals(1, splits.size());
125    WALInputFormat.WALSplit split = (WALInputFormat.WALSplit) splits.get(0);
126    assertEquals(archiveWal.toString(), split.getLogFileName());
127  }
128
129  @Test
130  public void testEmptyFileIsIgnoredWhenConfigured() throws IOException, InterruptedException {
131    List<InputSplit> splits = getSplitsForEmptyFile(true);
132    assertTrue(splits.isEmpty(), "Empty file should be ignored when IGNORE_EMPTY_FILES is true");
133  }
134
135  @Test
136  public void testEmptyFileIsIncludedWhenNotIgnored() throws IOException, InterruptedException {
137    List<InputSplit> splits = getSplitsForEmptyFile(false);
138    assertEquals(1, splits.size(),
139      "Empty file should be included when IGNORE_EMPTY_FILES is false");
140  }
141
142  private List<InputSplit> getSplitsForEmptyFile(boolean ignoreEmptyFiles)
143    throws IOException, InterruptedException {
144    Configuration conf = new Configuration();
145    conf.setBoolean(WALPlayer.IGNORE_EMPTY_FILES, ignoreEmptyFiles);
146
147    JobContext jobContext = Mockito.mock(JobContext.class);
148    Mockito.when(jobContext.getConfiguration()).thenReturn(conf);
149
150    LocatedFileStatus emptyFile = Mockito.mock(LocatedFileStatus.class);
151    Mockito.when(emptyFile.getLen()).thenReturn(0L);
152    Mockito.when(emptyFile.getPath()).thenReturn(new Path("/empty.wal"));
153
154    WALInputFormat inputFormat = new WALInputFormat() {
155      @Override
156      Path[] getInputPaths(Configuration conf) {
157        return new Path[] { new Path("/input") };
158      }
159
160      @Override
161      List<FileStatus> getFiles(FileSystem fs, Path inputPath, long startTime, long endTime,
162        Configuration conf) {
163        return Collections.singletonList(emptyFile);
164      }
165    };
166
167    return inputFormat.getSplits(jobContext, "", "");
168  }
169
170  /**
171   * Test that an empty WAL file (which causes WALHeaderEOFException) is gracefully handled and
172   * skipped rather than causing the job to fail.
173   */
174  @Test
175  public void testHandlesEmptyWALFile() throws Exception {
176    Configuration conf = TEST_UTIL.getConfiguration();
177
178    // Create an empty WAL file
179    Path walRootDir = CommonFSUtils.getWALRootDir(conf);
180    Path emptyWalFile =
181      new Path(walRootDir, "WALs/empty-wal-test/empty." + EnvironmentEdgeManager.currentTime());
182    TEST_UTIL.getTestFileSystem().mkdirs(emptyWalFile.getParent());
183    TEST_UTIL.getTestFileSystem().create(emptyWalFile).close();
184
185    try {
186      JobContext ctx = Mockito.mock(JobContext.class);
187      conf.set(FileInputFormat.INPUT_DIR, emptyWalFile.toString());
188      conf.set(WALPlayer.INPUT_FILES_SEPARATOR_KEY, ";");
189      Mockito.when(ctx.getConfiguration()).thenReturn(conf);
190      Job job = Job.getInstance(conf);
191      TableMapReduceUtil.initCredentialsForCluster(job, conf);
192      Mockito.when(ctx.getCredentials()).thenReturn(job.getCredentials());
193
194      // Create record reader and verify it handles the empty file gracefully
195      try (WALInputFormat.WALKeyRecordReader reader = new WALInputFormat.WALKeyRecordReader()) {
196        TaskAttemptContext taskCtx = Mockito.mock(TaskAttemptContext.class);
197        Mockito.when(taskCtx.getConfiguration()).thenReturn(conf);
198
199        WALInputFormat wif = new WALInputFormat();
200        List<InputSplit> splits = wif.getSplits(ctx);
201        assertEquals(1, splits.size());
202        WALInputFormat.WALSplit split = (WALInputFormat.WALSplit) splits.get(0);
203
204        // This should not throw WALHeaderEOFException - it should return false for nextKeyValue()
205        reader.initialize(split, taskCtx);
206        // nextKeyValue() should return false since the file is empty (reader is null)
207        assertFalse(reader.nextKeyValue());
208      }
209    } finally {
210      TEST_UTIL.getTestFileSystem().delete(emptyWalFile.getParent(), true);
211    }
212  }
213
214}