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.assertFalse;
022import static org.junit.jupiter.api.Assertions.assertTrue;
023
024import java.io.BufferedReader;
025import java.io.IOException;
026import java.io.InputStreamReader;
027import java.nio.charset.StandardCharsets;
028import java.util.Arrays;
029import java.util.List;
030import java.util.NavigableMap;
031import java.util.TreeMap;
032import org.apache.hadoop.conf.Configuration;
033import org.apache.hadoop.fs.FSDataInputStream;
034import org.apache.hadoop.fs.FileStatus;
035import org.apache.hadoop.fs.FileSystem;
036import org.apache.hadoop.fs.Path;
037import org.apache.hadoop.hbase.Cell;
038import org.apache.hadoop.hbase.CellBuilderFactory;
039import org.apache.hadoop.hbase.CellBuilderType;
040import org.apache.hadoop.hbase.HBaseTestingUtil;
041import org.apache.hadoop.hbase.TableName;
042import org.apache.hadoop.hbase.backup.util.BulkLoadProcessor;
043import org.apache.hadoop.hbase.client.RegionInfo;
044import org.apache.hadoop.hbase.client.RegionInfoBuilder;
045import org.apache.hadoop.hbase.io.asyncfs.monitor.StreamSlowMonitor;
046import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
047import org.apache.hadoop.hbase.regionserver.wal.ProtobufLogWriter;
048import org.apache.hadoop.hbase.regionserver.wal.WALUtil;
049import org.apache.hadoop.hbase.testclassification.LargeTests;
050import org.apache.hadoop.hbase.testclassification.MapReduceTests;
051import org.apache.hadoop.hbase.util.Bytes;
052import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
053import org.apache.hadoop.hbase.wal.FSHLogProvider;
054import org.apache.hadoop.hbase.wal.WAL;
055import org.apache.hadoop.hbase.wal.WALEdit;
056import org.apache.hadoop.hbase.wal.WALKeyImpl;
057import org.apache.hadoop.util.ToolRunner;
058import org.junit.jupiter.api.AfterAll;
059import org.junit.jupiter.api.BeforeAll;
060import org.junit.jupiter.api.BeforeEach;
061import org.junit.jupiter.api.Tag;
062import org.junit.jupiter.api.Test;
063import org.slf4j.Logger;
064import org.slf4j.LoggerFactory;
065
066import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
067
068import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
069import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos;
070
071/**
072 * Integration-like unit test for BulkLoadCollectorJob.
073 * <p>
074 * - Creates a WAL with a BULK_LOAD descriptor (ProtobufLogWriter).
075 * <p>
076 * - Runs BulkLoadCollectorJob.
077 * <p>
078 * - Verifies the job emits the expected store-file path.
079 */
080@Tag(MapReduceTests.TAG)
081@Tag(LargeTests.TAG)
082public class TestBulkLoadCollectorJobIntegration {
083
084  private static final Logger LOG =
085    LoggerFactory.getLogger(TestBulkLoadCollectorJobIntegration.class);
086  private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
087  private static Configuration conf;
088  private static FileSystem fs;
089  private static Path hbaseDir;
090  static final TableName tableName = TableName.valueOf(getName());
091  static final RegionInfo info = RegionInfoBuilder.newBuilder(tableName).build();
092  private static final byte[] family = Bytes.toBytes("column");
093  private static Path logDir;
094  protected MultiVersionConcurrencyControl mvcc;
095  protected static NavigableMap<byte[], Integer> scopes = new TreeMap<>(Bytes.BYTES_COMPARATOR);
096
097  private static String getName() {
098    return "TestBulkLoadCollectorJobIntegration";
099  }
100
101  @BeforeEach
102  public void setUp() throws Exception {
103    if (hbaseDir != null && fs != null) fs.delete(hbaseDir, true);
104    mvcc = new MultiVersionConcurrencyControl();
105  }
106
107  @BeforeAll
108  public static void setUpBeforeClass() throws Exception {
109    conf = TEST_UTIL.getConfiguration();
110    conf.setInt("dfs.blocksize", 1024 * 1024);
111    conf.setInt("dfs.replication", 1);
112
113    // Start a mini DFS cluster
114    TEST_UTIL.startMiniDFSCluster(3);
115
116    conf = TEST_UTIL.getConfiguration();
117    fs = TEST_UTIL.getDFSCluster().getFileSystem();
118
119    hbaseDir = TEST_UTIL.createRootDir();
120
121    // Use a deterministic test WAL directory under the test filesystem
122    logDir = new Path(TEST_UTIL.getDataTestDirOnTestFS(), "WALs/23-11-2024");
123    fs.mkdirs(logDir);
124  }
125
126  @AfterAll
127  public static void tearDownAfterClass() throws Exception {
128    if (fs != null && hbaseDir != null) fs.delete(hbaseDir, true);
129    TEST_UTIL.shutdownMiniDFSCluster();
130  }
131
132  /**
133   * Test that BulkLoadCollectorJob discovers and emits store-file paths from WAL files created
134   * using WALFactory (no RegionServer needed).
135   */
136  @Test
137  public void testBulkLoadCollectorEmitsStoreFilesFromWAL() throws Exception {
138    // Create WAL entry with BULK_LOAD descriptor
139    final String storeFileName = "storefile-abc.hfile";
140    WAL.Entry entry =
141      createBulkLoadWalEntry(info.getEncodedName(), Bytes.toString(family), storeFileName);
142
143    // Verify the processor would extract relative paths
144    List<Path> relativePaths =
145      BulkLoadProcessor.processBulkLoadFiles(entry.getKey(), entry.getEdit());
146    LOG.debug("BulkLoadProcessor returned {} relative path(s): {}", relativePaths.size(),
147      relativePaths);
148    assertEquals(1, relativePaths.size(),
149      "Expected exactly one relative path from BulkLoadProcessor");
150
151    // Build WAL file path and write WAL using ProtobufLogWriter into logDir
152    String walFileName = "wal-" + EnvironmentEdgeManager.currentTime();
153    Path walFilePath = new Path(logDir, walFileName);
154    fs.mkdirs(logDir);
155
156    FSHLogProvider.Writer writer = null;
157    try {
158      writer = new ProtobufLogWriter();
159      long blockSize = WALUtil.getWALBlockSize(conf, fs, walFilePath);
160      writer.init(fs, walFilePath, conf, true, blockSize,
161        StreamSlowMonitor.create(conf, walFileName));
162      writer.append(entry);
163      writer.sync(true);
164      writer.close();
165    } catch (Exception e) {
166      throw new IOException("Failed to write WAL via ProtobufLogWriter", e);
167    } finally {
168      try {
169        if (writer != null) writer.close();
170      } catch (Exception ignore) {
171      }
172    }
173
174    // Assert WAL file exists and has content
175    boolean exists = fs.exists(walFilePath);
176    long len = exists ? fs.getFileStatus(walFilePath).getLen() : -1L;
177    assertTrue(exists, "WAL file should exist at " + walFilePath);
178    assertTrue(len > 0, "WAL file should have non-zero length, actual=" + len);
179
180    // Run the MR job
181    Path walInputDir = logDir;
182    Path outDir = new Path("/tmp/test-bulk-files-output-" + System.currentTimeMillis());
183
184    int res = ToolRunner.run(TEST_UTIL.getConfiguration(),
185      new BulkLoadCollectorJob(TEST_UTIL.getConfiguration()),
186      new String[] { walInputDir.toString(), outDir.toString() });
187    assertEquals(0, res, "BulkLoadCollectorJob should exit with code 0");
188
189    // Inspect job output
190    FileSystem dfs = TEST_UTIL.getDFSCluster().getFileSystem();
191    assertTrue(dfs.exists(outDir), "Output directory should exist");
192
193    List<Path> partFiles = Arrays.stream(dfs.listStatus(outDir)).map(FileStatus::getPath)
194      .filter(p -> p.getName().startsWith("part-")).toList();
195
196    assertFalse(partFiles.isEmpty(), "Expect at least one part file in output");
197
198    // Read all lines (collect while stream is open)
199    List<String> lines = partFiles.stream().flatMap(p -> {
200      try (FSDataInputStream in = dfs.open(p);
201        BufferedReader r = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) {
202        List<String> fileLines = r.lines().toList();
203        return fileLines.stream();
204      } catch (Exception e) {
205        throw new RuntimeException(e);
206      }
207    }).toList();
208
209    assertFalse(lines.isEmpty(), "Job should have emitted at least one storefile path");
210
211    boolean found = lines.stream().anyMatch(l -> l.contains(storeFileName));
212    assertTrue(found,
213      "Expected emitted path to contain store file name: " + storeFileName + " ; got: " + lines);
214
215    // cleanup
216    dfs.delete(outDir, true);
217  }
218
219  private WAL.Entry createBulkLoadWalEntry(String regionName, String family, String... storeFiles) {
220
221    WALProtos.StoreDescriptor.Builder storeDescBuilder =
222      WALProtos.StoreDescriptor.newBuilder().setFamilyName(ByteString.copyFromUtf8(family))
223        .setStoreHomeDir(family).addAllStoreFile(Arrays.asList(storeFiles));
224
225    WALProtos.BulkLoadDescriptor.Builder bulkDescBuilder = WALProtos.BulkLoadDescriptor.newBuilder()
226      .setReplicate(true).setEncodedRegionName(ByteString.copyFromUtf8(regionName))
227      .setTableName(ProtobufUtil.toProtoTableName(tableName)).setBulkloadSeqNum(1000)
228      .addStores(storeDescBuilder);
229
230    byte[] valueBytes = bulkDescBuilder.build().toByteArray();
231
232    WALEdit edit = new WALEdit();
233    Cell cell = CellBuilderFactory.create(CellBuilderType.DEEP_COPY).setType(Cell.Type.Put)
234      .setRow(new byte[] { 1 }).setFamily(WALEdit.METAFAMILY).setQualifier(WALEdit.BULK_LOAD)
235      .setValue(valueBytes).build();
236    edit.add(cell);
237
238    long ts = EnvironmentEdgeManager.currentTime();
239    WALKeyImpl key = getWalKeyImpl(ts, scopes);
240    return new WAL.Entry(key, edit);
241  }
242
243  protected WALKeyImpl getWalKeyImpl(final long time, NavigableMap<byte[], Integer> scopes) {
244    return new WALKeyImpl(info.getEncodedNameAsBytes(), tableName, time, mvcc, scopes);
245  }
246}