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.backup.mapreduce;
019
020import static org.junit.jupiter.api.Assertions.assertEquals;
021import static org.junit.jupiter.api.Assertions.assertThrows;
022import static org.mockito.ArgumentMatchers.any;
023import static org.mockito.ArgumentMatchers.eq;
024import static org.mockito.Mockito.mock;
025import static org.mockito.Mockito.never;
026import static org.mockito.Mockito.times;
027import static org.mockito.Mockito.verify;
028import static org.mockito.Mockito.when;
029
030import java.io.IOException;
031import java.text.SimpleDateFormat;
032import java.util.Arrays;
033import java.util.Collections;
034import org.apache.hadoop.conf.Configuration;
035import org.apache.hadoop.fs.Path;
036import org.apache.hadoop.hbase.HBaseConfiguration;
037import org.apache.hadoop.hbase.TableName;
038import org.apache.hadoop.hbase.backup.util.BulkLoadProcessor;
039import org.apache.hadoop.hbase.mapreduce.WALInputFormat;
040import org.apache.hadoop.hbase.mapreduce.WALPlayer;
041import org.apache.hadoop.hbase.testclassification.LargeTests;
042import org.apache.hadoop.hbase.testclassification.MapReduceTests;
043import org.apache.hadoop.hbase.wal.WALEdit;
044import org.apache.hadoop.hbase.wal.WALKey;
045import org.apache.hadoop.io.NullWritable;
046import org.apache.hadoop.io.Text;
047import org.apache.hadoop.mapreduce.Job;
048import org.apache.hadoop.mapreduce.Mapper;
049import org.apache.hadoop.mapreduce.Reducer;
050import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
051import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
052import org.junit.jupiter.api.AfterEach;
053import org.junit.jupiter.api.BeforeEach;
054import org.junit.jupiter.api.Tag;
055import org.junit.jupiter.api.Test;
056import org.mockito.MockedStatic;
057import org.mockito.Mockito;
058
059/**
060 * Unit tests for BulkLoadCollectorJob (mapper, reducer and job creation/validation).
061 */
062@Tag(MapReduceTests.TAG)
063@Tag(LargeTests.TAG)
064public class TestBulkLoadCollectorJob {
065
066  private Configuration conf;
067
068  @BeforeEach
069  public void setUp() {
070    // fresh configuration for each test
071    conf = HBaseConfiguration.create();
072  }
073
074  @AfterEach
075  public void tearDown() {
076    // nothing for now
077  }
078
079  /**
080   * Ensures {@link BulkLoadCollectorJob#createSubmittableJob(String[])} correctly configures
081   * input/output paths and parses time options into the job configuration.
082   */
083  @Test
084  public void testCreateSubmittableJobValid() throws Exception {
085    // set a start time option to make sure setupTime runs and applies it
086    String dateStr = "2001-02-20T16:35:06.99";
087    conf.set(WALInputFormat.START_TIME_KEY, dateStr);
088
089    BulkLoadCollectorJob jobDriver = new BulkLoadCollectorJob(conf);
090    String inputDirs = new Path("file:/wals/input").toString();
091    String outDir = new Path("file:/out/bulk").toString();
092    Job job = jobDriver.createSubmittableJob(new String[] { inputDirs, outDir });
093
094    // Input path set
095    Path[] inPaths = FileInputFormat.getInputPaths(job);
096    assertEquals(1, inPaths.length);
097    assertEquals(inputDirs, inPaths[0].toString());
098
099    // Output path set
100    Path out = FileOutputFormat.getOutputPath(job);
101    assertEquals(new Path(outDir), out);
102
103    // Ensure the conf had START_TIME_KEY parsed to a long (setupTime executed)
104    long parsed = conf.getLong(WALInputFormat.START_TIME_KEY, -1L);
105    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SS");
106    long expected = sdf.parse(dateStr).getTime();
107    assertEquals(expected, parsed);
108  }
109
110  /**
111   * Verifies that {@link BulkLoadCollectorJob#createSubmittableJob(String[])} throws an IOException
112   * when called with insufficient or null arguments.
113   */
114  @Test
115  public void testCreateSubmittableJob_throwsForInsufficientArgs() throws Exception {
116    BulkLoadCollectorJob jobDriver = new BulkLoadCollectorJob(conf);
117    // this call must throw IOException for the test to pass
118    assertThrows(IOException.class,
119      () -> jobDriver.createSubmittableJob(new String[] { "file:/only/one/arg" }));
120  }
121
122  @Test
123  public void testCreateSubmittableJob_throwsForNullArgs() throws Exception {
124    BulkLoadCollectorJob jobDriver = new BulkLoadCollectorJob(conf);
125    // this call must throw IOException for the test to pass
126    assertThrows(IOException.class, () -> jobDriver.createSubmittableJob(null));
127  }
128
129  /**
130   * Verifies that {@link BulkLoadCollectorJob.BulkLoadCollectorMapper} ignores WAL entries whose
131   * table is not present in the configured tables map.
132   */
133  @Test
134  public void testMapperIgnoresWhenTableNotInMap() throws Exception {
135    // Prepare mapper and a mocked MapReduce context
136    BulkLoadCollectorJob.BulkLoadCollectorMapper mapper =
137      new BulkLoadCollectorJob.BulkLoadCollectorMapper();
138    @SuppressWarnings("unchecked")
139    Mapper<WALKey, WALEdit, Text, NullWritable>.Context ctx = mock(Mapper.Context.class);
140
141    // Build a Configuration that only allows a single table: ns:allowed
142    // Note: TABLES_KEY / TABLE_MAP_KEY are the same constants used by the mapper.setup(...)
143    Configuration cfgForTest = new Configuration(conf);
144    cfgForTest.setStrings(WALPlayer.TABLES_KEY, "ns:allowed");
145    cfgForTest.setStrings(WALPlayer.TABLE_MAP_KEY, "ns:allowed"); // maps to itself
146
147    // Have the mocked context return our test configuration when mapper.setup() runs
148    when(ctx.getConfiguration()).thenReturn(cfgForTest);
149    mapper.setup(ctx);
150
151    // Create a WALKey for a table that is NOT in the allowed map (ns:other)
152    WALKey keyForOtherTable = mock(WALKey.class);
153    when(keyForOtherTable.getTableName()).thenReturn(TableName.valueOf("ns:other"));
154    WALEdit walEdit = mock(WALEdit.class);
155
156    // Static-mock BulkLoadProcessor to ensure it would not be relied on:
157    // even if invoked unexpectedly, it returns a non-empty list, but we will assert no writes
158    // occurred.
159    try (MockedStatic<BulkLoadProcessor> proc = Mockito.mockStatic(BulkLoadProcessor.class)) {
160      proc.when(() -> BulkLoadProcessor.processBulkLoadFiles(any(), any()))
161        .thenReturn(Collections.singletonList(new Path("x")));
162
163      // Invoke mapper - because the table is not allowed, mapper should do nothing
164      mapper.map(keyForOtherTable, walEdit, ctx);
165
166      // Assert: mapper did not write any output to the context
167      verify(ctx, never()).write(any(Text.class), any(NullWritable.class));
168    }
169  }
170
171  /**
172   * Verifies that {@link BulkLoadCollectorJob.BulkLoadCollectorMapper} safely handles null inputs.
173   * <p>
174   * The mapper should ignore WAL entries when either the WAL key or the WALEdit value is null, and
175   * must not emit any output in those cases.
176   * </p>
177   * @throws Exception on test failure
178   */
179  @Test
180  public void testMapperHandlesNullKeyOrValue() throws Exception {
181    BulkLoadCollectorJob.BulkLoadCollectorMapper mapper =
182      new BulkLoadCollectorJob.BulkLoadCollectorMapper();
183    @SuppressWarnings("unchecked")
184    Mapper<WALKey, WALEdit, Text, NullWritable>.Context ctx = mock(Mapper.Context.class);
185    when(ctx.getConfiguration()).thenReturn(conf);
186    mapper.setup(ctx);
187
188    // null key
189    mapper.map(null, mock(WALEdit.class), ctx);
190    // null value
191    mapper.map(mock(WALKey.class), null, ctx);
192
193    // ensure no writes
194    verify(ctx, never()).write(any(Text.class), any(NullWritable.class));
195  }
196
197  /**
198   * Verifies that {@link BulkLoadCollectorJob.DedupReducer} writes each unique key exactly once.
199   */
200  @Test
201  public void testDedupReducerWritesOnce() throws Exception {
202    BulkLoadCollectorJob.DedupReducer reducer = new BulkLoadCollectorJob.DedupReducer();
203    @SuppressWarnings("unchecked")
204    Reducer<Text, NullWritable, Text, NullWritable>.Context ctx = mock(Reducer.Context.class);
205
206    Text key = new Text("/some/path");
207
208    // Simulate three duplicate values for the same key; reducer should still write the key once.
209    Iterable<NullWritable> vals =
210      Arrays.asList(NullWritable.get(), NullWritable.get(), NullWritable.get());
211
212    reducer.reduce(key, vals, ctx);
213
214    // verify exactly once write with the same key
215    verify(ctx, times(1)).write(eq(key), eq(NullWritable.get()));
216  }
217}