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.util;
019
020import java.io.BufferedReader;
021import java.io.IOException;
022import java.io.InputStreamReader;
023import java.nio.charset.StandardCharsets;
024import java.util.ArrayList;
025import java.util.Arrays;
026import java.util.LinkedHashSet;
027import java.util.List;
028import java.util.Set;
029import org.apache.hadoop.conf.Configuration;
030import org.apache.hadoop.fs.FSDataInputStream;
031import org.apache.hadoop.fs.FileSystem;
032import org.apache.hadoop.fs.LocatedFileStatus;
033import org.apache.hadoop.fs.Path;
034import org.apache.hadoop.fs.RemoteIterator;
035import org.apache.hadoop.hbase.TableName;
036import org.apache.hadoop.hbase.backup.mapreduce.BulkLoadCollectorJob;
037import org.apache.hadoop.hbase.mapreduce.WALInputFormat;
038import org.apache.hadoop.hbase.mapreduce.WALPlayer;
039import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
040import org.apache.hadoop.util.Tool;
041import org.apache.yetus.audience.InterfaceAudience;
042import org.slf4j.Logger;
043import org.slf4j.LoggerFactory;
044
045/**
046 * Utility to run BulkLoadCollectorJob over a comma-separated list of WAL directories and return a
047 * deduplicated list of discovered bulk-load file paths.
048 */
049@InterfaceAudience.Private
050public final class BulkFilesCollector {
051
052  private static final Logger LOG = LoggerFactory.getLogger(BulkFilesCollector.class);
053
054  private BulkFilesCollector() {
055    /* static only */ }
056
057  /**
058   * Convenience overload: collector will create and configure BulkLoadCollectorJob internally.
059   * @param conf           cluster/configuration used to initialize job and access FS
060   * @param walDirsCsv     comma-separated WAL directories
061   * @param restoreRootDir parent path under which temporary output dir will be created
062   * @param sourceTable    source table name (for args/logging)
063   * @param targetTable    target table name (for args/logging)
064   * @param startTime      start time (ms) to set in the job config (WALInputFormat.START_TIME_KEY)
065   * @param endTime        end time (ms) to set in the job config (WALInputFormat.END_TIME_KEY)
066   * @return deduplicated list of Paths discovered by the collector
067   * @throws IOException on IO or job failure
068   */
069  public static List<Path> collectFromWalDirs(Configuration conf, String walDirsCsv,
070    Path restoreRootDir, TableName sourceTable, TableName targetTable, long startTime, long endTime)
071    throws IOException {
072
073    // prepare job Tool
074    Configuration jobConf = new Configuration(conf);
075    if (startTime > 0) jobConf.setLong(WALInputFormat.START_TIME_KEY, startTime);
076    if (endTime > 0) jobConf.setLong(WALInputFormat.END_TIME_KEY, endTime);
077
078    // ignore empty WAL files by default to make collection robust
079    jobConf.setBoolean(WALPlayer.IGNORE_EMPTY_FILES, true);
080
081    BulkLoadCollectorJob bulkCollector = new BulkLoadCollectorJob();
082    bulkCollector.setConf(jobConf);
083
084    return collectFromWalDirs(conf, walDirsCsv, restoreRootDir, sourceTable, targetTable,
085      bulkCollector);
086  }
087
088  /**
089   * Primary implementation: runs the provided Tool (BulkLoadCollectorJob) with args "<walDirsCsv>
090   * <bulkFilesOut> <sourceTable> <targetTable>" and returns deduped list of Paths.
091   */
092  public static List<Path> collectFromWalDirs(Configuration conf, String walDirsCsv,
093    Path restoreRootDir, TableName sourceTable, TableName targetTable, Tool bulkCollector)
094    throws IOException {
095
096    if (walDirsCsv == null || walDirsCsv.trim().isEmpty()) {
097      throw new IOException(
098        "walDirsCsv must be a non-empty comma-separated list of WAL directories");
099    }
100
101    List<String> walDirs =
102      Arrays.stream(walDirsCsv.split(",")).map(String::trim).filter(s -> !s.isEmpty()).toList();
103
104    if (walDirs.isEmpty()) {
105      throw new IOException("walDirsCsv did not contain any entries: '" + walDirsCsv + "'");
106    }
107
108    List<String> existing = new ArrayList<>();
109    for (String d : walDirs) {
110      Path p = new Path(d);
111      try {
112        FileSystem fsForPath = p.getFileSystem(conf);
113        if (fsForPath.exists(p)) {
114          existing.add(d);
115        } else {
116          LOG.debug("WAL dir does not exist: {}", d);
117        }
118      } catch (IOException e) {
119        // If getting FS or checking existence fails, treat as missing but log the cause.
120        LOG.warn("Error checking WAL dir {}: {}", d, e.toString());
121      }
122    }
123
124    // If any of the provided walDirs are missing, fail with an informative message.
125    List<String> missing = new ArrayList<>(walDirs);
126    missing.removeAll(existing);
127
128    if (!missing.isEmpty()) {
129      throw new IOException(
130        "Some of the provided WAL paths do not exist: " + String.join(", ", missing));
131    }
132
133    // Create unique temporary output dir under restoreRootDir, e.g.
134    // <restoreRootDir>/_wal_collect_<table_name><ts>
135    final String unique = String.format("_wal_collect_%s%d", sourceTable.getQualifierAsString(),
136      EnvironmentEdgeManager.currentTime());
137    final Path bulkFilesOut = new Path(restoreRootDir, unique);
138
139    FileSystem fs = bulkFilesOut.getFileSystem(conf);
140
141    try {
142      // If bulkFilesOut exists for some reason, delete it.
143      if (fs.exists(bulkFilesOut)) {
144        LOG.info("Temporary bulkload file collect output directory {} already exists - deleting.",
145          bulkFilesOut);
146        fs.delete(bulkFilesOut, true);
147      }
148
149      final String[] args = new String[] { walDirsCsv, bulkFilesOut.toString(),
150        sourceTable.getNameAsString(), targetTable.getNameAsString() };
151
152      LOG.info("Running bulk collector Tool with args: {}", (Object) args);
153
154      int exitCode;
155      try {
156        exitCode = bulkCollector.run(args);
157      } catch (Exception e) {
158        LOG.error("Error during BulkLoadCollectorJob for {}: {}", sourceTable, e.getMessage(), e);
159        throw new IOException("Exception during BulkLoadCollectorJob collect", e);
160      }
161
162      if (exitCode != 0) {
163        throw new IOException("Bulk collector Tool returned non-zero exit code: " + exitCode);
164      }
165
166      LOG.info("BulkLoadCollectorJob collect completed successfully for {}", sourceTable);
167
168      // read and dedupe
169      List<Path> results = readBulkFilesListFromOutput(fs, bulkFilesOut);
170      LOG.info("BulkFilesCollector: discovered {} unique bulk-load files", results.size());
171      return results;
172    } finally {
173      // best-effort cleanup
174      try {
175        if (fs.exists(bulkFilesOut)) {
176          boolean deleted = fs.delete(bulkFilesOut, true);
177          if (!deleted) {
178            LOG.warn("Could not delete temporary bulkFilesOut directory {}", bulkFilesOut);
179          } else {
180            LOG.debug("Deleted temporary bulkFilesOut directory {}", bulkFilesOut);
181          }
182        }
183      } catch (IOException ioe) {
184        LOG.warn("Exception while deleting temporary bulkload file collect output dir {}: {}",
185          bulkFilesOut, ioe.getMessage(), ioe);
186      }
187    }
188  }
189
190  // reads all non-hidden files under bulkFilesOut, collects lines in insertion order, returns Paths
191  private static List<Path> readBulkFilesListFromOutput(FileSystem fs, Path bulkFilesOut)
192    throws IOException {
193    if (!fs.exists(bulkFilesOut)) {
194      LOG.warn("BulkFilesCollector: bulkFilesOut directory does not exist: {}", bulkFilesOut);
195      return new ArrayList<>();
196    }
197
198    RemoteIterator<LocatedFileStatus> it = fs.listFiles(bulkFilesOut, true);
199    Set<String> dedupe = new LinkedHashSet<>();
200
201    while (it.hasNext()) {
202      LocatedFileStatus status = it.next();
203      Path p = status.getPath();
204      String name = p.getName();
205      // skip hidden/system files like _SUCCESS or _logs
206      if (name.startsWith("_") || name.startsWith(".")) continue;
207
208      try (FSDataInputStream in = fs.open(p);
209        BufferedReader br = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))) {
210        String line;
211        while ((line = br.readLine()) != null) {
212          line = line.trim();
213          if (line.isEmpty()) continue;
214          dedupe.add(line);
215        }
216      }
217    }
218
219    List<Path> result = new ArrayList<>(dedupe.size());
220    for (String s : dedupe)
221      result.add(new Path(s));
222
223    LOG.info("Collected {} unique bulk-load store files.", result.size());
224    return result;
225  }
226}