working replication endpoint with unit tests

Author: vinayakphegde

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/replication/ContinuousBackupReplicationEndpoint.java

@@ -42,12 +42,19 @@
 import org.apache.hadoop.hbase.replication.ReplicationResult;
 import org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceInterface;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.wal.FSHLogProvider;
 import org.apache.hadoop.hbase.wal.WAL;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * ContinuousBackupReplicationEndpoint is responsible for replicating WAL entries to a backup
+ * storage. It organizes WAL entries by day and periodically flushes the data, ensuring that WAL
+ * files do not exceed the configured size. The class includes mechanisms for handling the WAL
+ * files, performing bulk load backups, and ensuring that the replication process is safe.
+ */
 @InterfaceAudience.Private
 public class ContinuousBackupReplicationEndpoint extends BaseReplicationEndpoint {
   private static final Logger LOG =
@@ -75,9 +82,9 @@ public class ContinuousBackupReplicationEndpoint extends BaseReplicationEndpoint
   private String peerId;
   private ScheduledExecutorService flushExecutor;
 
-  private static final long ONE_DAY_IN_MILLISECONDS = TimeUnit.DAYS.toMillis(1);
+  public static final long ONE_DAY_IN_MILLISECONDS = TimeUnit.DAYS.toMillis(1);
   public static final String WAL_FILE_PREFIX = "wal_file.";
-  private static final String DATE_FORMAT = "yyyy-MM-dd";
+  public static final String DATE_FORMAT = "yyyy-MM-dd";
 
   @Override
   public void init(Context context) throws IOException {
@@ -139,6 +146,8 @@ private void flushAndBackupSafely() {
       LOG.info("{} Periodic WAL flush triggered", Utils.logPeerId(peerId));
       flushWriters();
       replicationSource.persistOffsets();
+      LOG.info("{} Periodic WAL flush and offset persistence completed successfully",
+        Utils.logPeerId(peerId));
     } catch (IOException e) {
       LOG.error("{} Error during WAL flush: {}", Utils.logPeerId(peerId), e.getMessage(), e);
     } finally {
@@ -150,7 +159,13 @@ private void flushWriters() throws IOException {
     for (Map.Entry<Long, FSHLogProvider.Writer> entry : walWriters.entrySet()) {
       FSHLogProvider.Writer writer = entry.getValue();
       if (writer != null) {
-        writer.close();
+        try {
+          writer.close();
+        } catch (IOException e) {
+          LOG.error("{} Failed to close WAL writer for day: {}. Error: {}", Utils.logPeerId(peerId),
+            entry.getKey(), e.getMessage());
+          throw e;
+        }
       }
     }
     walWriters.clear();
@@ -250,14 +265,17 @@ private FSHLogProvider.Writer createWalWriter(long dayInMillis) {
       fs.mkdirs(dayDir);
 
       // Generate a unique WAL file name
-      String walFileName = WAL_FILE_PREFIX + dayInMillis + "." + UUID.randomUUID();
+      long currentTime = EnvironmentEdgeManager.getDelegate().currentTime();
+      String walFileName = WAL_FILE_PREFIX + currentTime + "." + UUID.randomUUID();
       Path walFilePath = new Path(dayDir, walFileName);
 
       // Initialize the WAL writer
       FSHLogProvider.Writer writer =
         ObjectStoreProtobufWalWriter.class.getDeclaredConstructor().newInstance();
       writer.init(fs, walFilePath, conf, true, WALUtil.getWALBlockSize(conf, fs, walFilePath),
         StreamSlowMonitor.create(conf, walFileName));
+
+      LOG.debug("{} Created WAL writer for day: {}", Utils.logPeerId(peerId), dayDirectoryName);
       return writer;
     } catch (Exception e) {
       throw new UncheckedIOException(
@@ -282,12 +300,15 @@ protected void doStop() {
 
   private void close() {
     shutdownFlushExecutor();
+    lock.lock();
     try {
       flushWriters();
       replicationSource.persistOffsets();
     } catch (IOException e) {
       LOG.error("{} Failed to Flush Open Wal Writers: {}", Utils.logPeerId(peerId), e.getMessage(),
         e);
+    } finally {
+      lock.unlock();
     }
   }
 
@@ -302,13 +323,14 @@ private void uploadBulkLoadFiles(List<Path> bulkLoadFiles) throws IOException {
         LOG.info("{} Bulk load file {} successfully backed up to {}", Utils.logPeerId(peerId), file,
           destPath);
       } catch (IOException e) {
-        LOG.error("{} Failed to back up bulk load file: {}", Utils.logPeerId(peerId), file, e);
+        LOG.error("{} Failed to back up bulk load file {}: {}", Utils.logPeerId(peerId), file,
+          e.getMessage(), e);
         throw e;
       }
     }
   }
 
-  public Path getBulkLoadFileStagingPath(Path relativePathFromNamespace) throws IOException {
+  private Path getBulkLoadFileStagingPath(Path relativePathFromNamespace) throws IOException {
     FileSystem rootFs = CommonFSUtils.getRootDirFileSystem(conf);
     Path rootDir = CommonFSUtils.getRootDir(conf);
     Path baseNSDir = new Path(HConstants.BASE_NAMESPACE_DIR);
@@ -344,6 +366,8 @@ private void shutdownFlushExecutor() {
           !flushExecutor.awaitTermination(EXECUTOR_TERMINATION_TIMEOUT_SECONDS, TimeUnit.SECONDS)
         ) {
           flushExecutor.shutdownNow();
+          LOG.warn("{} Flush executor did not terminate within timeout, forcing shutdown.",
+            Utils.logPeerId(peerId));
         }
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/replication/ObjectStoreProtobufWalWriter.java

@@ -18,55 +18,33 @@
 package org.apache.hadoop.hbase.backup.replication;
 
 import java.io.IOException;
-import java.io.OutputStream;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FSDataOutputStreamBuilder;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.ExtendedCell;
 import org.apache.hadoop.hbase.io.asyncfs.monitor.StreamSlowMonitor;
-import org.apache.hadoop.hbase.regionserver.wal.AbstractProtobufLogWriter;
+import org.apache.hadoop.hbase.regionserver.wal.ProtobufLogWriter;
 import org.apache.hadoop.hbase.util.AtomicUtils;
-import org.apache.hadoop.hbase.wal.FSHLogProvider;
-import org.apache.hadoop.hbase.wal.WAL;
+import org.apache.hadoop.hbase.util.CommonFSUtils;
 import org.apache.yetus.audience.InterfaceAudience;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos;
 
+/**
+ * A custom implementation of {@link ProtobufLogWriter} that provides support for writing
+ * protobuf-based WAL (Write-Ahead Log) entries to object store-backed files.
+ * <p>
+ * This class overrides the {@link ProtobufLogWriter#sync(boolean)} and
+ * {@link ProtobufLogWriter#initOutput(FileSystem, Path, boolean, int, short, long, StreamSlowMonitor, boolean)}
+ * methods to ensure compatibility with object stores, while ignoring specific capability checks
+ * such as HFLUSH and HSYNC. These checks are often not supported by some object stores, and
+ * bypassing them ensures smooth operation in such environments.
+ * </p>
+ */
 @InterfaceAudience.Private
-public class ObjectStoreProtobufWalWriter extends AbstractProtobufLogWriter
-  implements FSHLogProvider.Writer {
-  private static final Logger LOG = LoggerFactory.getLogger(ObjectStoreProtobufWalWriter.class);
+public class ObjectStoreProtobufWalWriter extends ProtobufLogWriter {
 
   protected FSDataOutputStream output;
   private final AtomicLong syncedLength = new AtomicLong(0);
 
-  @Override
-  public void append(WAL.Entry entry) throws IOException {
-    entry.getKey().getBuilder(compressor).setFollowingKvCount(entry.getEdit().size()).build()
-      .writeDelimitedTo(output);
-    for (Cell cell : entry.getEdit().getCells()) {
-      // cellEncoder must assume little about the stream, since we write PB and cells in turn.
-      cellEncoder.write((ExtendedCell) cell);
-    }
-    length.set(output.getPos());
-  }
-
-  @Override
-  public void close() throws IOException {
-    if (this.output != null) {
-      if (!trailerWritten) {
-        writeWALTrailer();
-      }
-      this.output.close();
-      this.output = null;
-    }
-  }
-
   @Override
   public void sync(boolean forceSync) throws IOException {
     FSDataOutputStream fsDataOutputstream = this.output;
@@ -78,49 +56,16 @@ public void sync(boolean forceSync) throws IOException {
   }
 
   @Override
-  public long getSyncedLength() {
-    return this.syncedLength.get();
-  }
-
-  @Override
-  protected void initOutput(FileSystem fs, Path path, boolean overridable, int bufferSize,
+  protected void initOutput(FileSystem fs, Path path, boolean overwritable, int bufferSize,
     short replication, long blockSize, StreamSlowMonitor monitor, boolean noLocalWrite)
     throws IOException {
-    FSDataOutputStreamBuilder<?, ?> builder = fs.createFile(path).overwrite(overridable)
-      .bufferSize(bufferSize).replication(replication).blockSize(blockSize);
-    this.output = builder.build();
-  }
-
-  @Override
-  protected void closeOutputIfNecessary() {
-    if (this.output != null) {
-      try {
-        this.output.close();
-      } catch (IOException e) {
-        LOG.warn("Close output failed", e);
-      }
+    try {
+      super.initOutput(fs, path, overwritable, bufferSize, replication, blockSize, monitor,
+        noLocalWrite);
+    } catch (CommonFSUtils.StreamLacksCapabilityException e) {
+      // Ignore capability check for HFLUSH and HSYNC capabilities
+      // Some object stores may not support these capabilities, so we bypass the exception handling
+      // to ensure compatibility with such stores.
     }
   }
-
-  @Override
-  protected long writeMagicAndWALHeader(byte[] magic, WALProtos.WALHeader header)
-    throws IOException {
-    output.write(magic);
-    header.writeDelimitedTo(output);
-    return output.getPos();
-  }
-
-  @Override
-  protected OutputStream getOutputStreamForCellEncoder() {
-    return this.output;
-  }
-
-  @Override
-  protected long writeWALTrailerAndMagic(WALProtos.WALTrailer trailer, byte[] magic)
-    throws IOException {
-    trailer.writeTo(output);
-    output.writeInt(trailer.getSerializedSize());
-    output.write(magic);
-    return output.getPos();
-  }
 }