add javadocs and log lines

Author: vinayakphegde

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

@@ -25,6 +25,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+/**
+ * Initializes and organizes backup directories for continuous Write-Ahead Logs (WALs) and
+ * bulk-loaded files within the specified backup root directory.
+ */
 @InterfaceAudience.Private
 public class BackupFileSystemManager {
   private static final Logger LOG = LoggerFactory.getLogger(BackupFileSystemManager.class);

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

@@ -96,6 +96,7 @@ public void init(Context context) throws IOException {
     initializePeerUUID();
     initializeBackupFileSystemManager();
     startWalFlushExecutor();
+    LOG.info("{} Initialization complete", Utils.logPeerId(peerId));
   }
 
   private void initializePeerUUID() throws IOException {
@@ -156,20 +157,24 @@ private void flushAndBackupSafely() {
   }
 
   private void flushWriters() throws IOException {
+    LOG.info("{} Flushing {} WAL writers", Utils.logPeerId(peerId), walWriters.size());
     for (Map.Entry<Long, FSHLogProvider.Writer> entry : walWriters.entrySet()) {
       FSHLogProvider.Writer writer = entry.getValue();
       if (writer != null) {
+        LOG.debug("{} Closing WAL writer for day: {}", Utils.logPeerId(peerId), entry.getKey());
         try {
           writer.close();
+          LOG.debug("{} Successfully closed WAL writer for day: {}", Utils.logPeerId(peerId),
+            entry.getKey());
         } catch (IOException e) {
           LOG.error("{} Failed to close WAL writer for day: {}. Error: {}", Utils.logPeerId(peerId),
-            entry.getKey(), e.getMessage());
+            entry.getKey(), e.getMessage(), e);
           throw e;
         }
       }
     }
     walWriters.clear();
-    LOG.debug("{} WAL writers flushed and cleared", Utils.logPeerId(peerId));
+    LOG.info("{} WAL writers flushed and cleared", Utils.logPeerId(peerId));
   }
 
   @Override
@@ -198,19 +203,30 @@ public ReplicationResult replicate(ReplicateContext replicateContext) {
       return ReplicationResult.SUBMITTED;
     }
 
+    LOG.debug("{} Received {} WAL entries for replication", Utils.logPeerId(peerId),
+      entries.size());
+
     Map<Long, List<WAL.Entry>> groupedEntries = groupEntriesByDay(entries);
-    lock.lock();
+    LOG.debug("{} Grouped WAL entries by day: {}", Utils.logPeerId(peerId),
+      groupedEntries.keySet());
 
+    lock.lock();
     try {
       for (Map.Entry<Long, List<WAL.Entry>> entry : groupedEntries.entrySet()) {
+        LOG.debug("{} Backing up {} WAL entries for day {}", Utils.logPeerId(peerId),
+          entry.getValue().size(), entry.getKey());
         backupWalEntries(entry.getKey(), entry.getValue());
       }
 
       if (isAnyWriterFull()) {
+        LOG.debug("{} Some WAL writers reached max size, triggering flush",
+          Utils.logPeerId(peerId));
         flushWriters();
+        LOG.debug("{} Replication committed after WAL flush", Utils.logPeerId(peerId));
         return ReplicationResult.COMMITTED;
       }
 
+      LOG.debug("{} Replication submitted successfully", Utils.logPeerId(peerId));
       return ReplicationResult.SUBMITTED;
     } catch (IOException e) {
       LOG.error("{} Replication failed. Error details: {}", Utils.logPeerId(peerId), e.getMessage(),
@@ -237,16 +253,29 @@ private boolean isWriterFull(FSHLogProvider.Writer writer) {
   }
 
   private void backupWalEntries(long day, List<WAL.Entry> walEntries) throws IOException {
+    LOG.debug("{} Starting backup of {} WAL entries for day {}", Utils.logPeerId(peerId),
+      walEntries.size(), day);
+
     try {
       FSHLogProvider.Writer walWriter = walWriters.computeIfAbsent(day, this::createWalWriter);
       List<Path> bulkLoadFiles = BulkLoadProcessor.processBulkLoadFiles(walEntries);
+
+      if (LOG.isTraceEnabled()) {
+        LOG.trace("{} Processed {} bulk load files for WAL entries", Utils.logPeerId(peerId),
+          bulkLoadFiles.size());
+        LOG.trace("{} Bulk load files: {}", Utils.logPeerId(peerId),
+          bulkLoadFiles.stream().map(Path::toString).collect(Collectors.joining(", ")));
+      }
+
       for (WAL.Entry entry : walEntries) {
         walWriter.append(entry);
       }
       walWriter.sync(true);
       uploadBulkLoadFiles(bulkLoadFiles);
     } catch (UncheckedIOException e) {
       String errorMsg = Utils.logPeerId(peerId) + " Failed to get or create WAL Writer for " + day;
+      LOG.error("{} Backup failed for day {}. Error: {}", Utils.logPeerId(peerId), day,
+        e.getMessage(), e);
       throw new IOException(errorMsg, e);
     }
   }
@@ -275,7 +304,7 @@ private FSHLogProvider.Writer createWalWriter(long dayInMillis) {
       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);
+      LOG.info("{} WAL writer created: {}", Utils.logPeerId(peerId), walFilePath);
       return writer;
     } catch (Exception e) {
       throw new UncheckedIOException(
@@ -299,6 +328,7 @@ protected void doStop() {
   }
 
   private void close() {
+    LOG.info("{} Closing WAL replication component...", Utils.logPeerId(peerId));
     shutdownFlushExecutor();
     lock.lock();
     try {
@@ -309,17 +339,29 @@ private void close() {
         e);
     } finally {
       lock.unlock();
+      LOG.info("{} WAL replication component closed.", Utils.logPeerId(peerId));
     }
   }
 
   private void uploadBulkLoadFiles(List<Path> bulkLoadFiles) throws IOException {
+    LOG.debug("{} Starting upload of {} bulk load files", Utils.logPeerId(peerId),
+      bulkLoadFiles.size());
+
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("{} Bulk load files to upload: {}", Utils.logPeerId(peerId),
+        bulkLoadFiles.stream().map(Path::toString).collect(Collectors.joining(", ")));
+    }
     for (Path file : bulkLoadFiles) {
       Path sourcePath = getBulkLoadFileStagingPath(file);
       Path destPath = new Path(backupFileSystemManager.getBulkLoadFilesDir(), file);
 
       try {
+        LOG.debug("{} Copying bulk load file from {} to {}", Utils.logPeerId(peerId), sourcePath,
+          destPath);
+
         FileUtil.copy(CommonFSUtils.getRootDirFileSystem(conf), sourcePath,
           backupFileSystemManager.getBackupFs(), destPath, false, conf);
+
         LOG.info("{} Bulk load file {} successfully backed up to {}", Utils.logPeerId(peerId), file,
           destPath);
       } catch (IOException e) {
@@ -328,6 +370,8 @@ private void uploadBulkLoadFiles(List<Path> bulkLoadFiles) throws IOException {
         throw e;
       }
     }
+
+    LOG.debug("{} Completed upload of bulk load files", Utils.logPeerId(peerId));
   }
 
   private Path getBulkLoadFileStagingPath(Path relativePathFromNamespace) throws IOException {
@@ -338,20 +382,34 @@ private Path getBulkLoadFileStagingPath(Path relativePathFromNamespace) throws I
     Path hFileArchiveDir =
       new Path(rootDir, new Path(HConstants.HFILE_ARCHIVE_DIRECTORY, baseNSDir));
 
+    LOG.debug("{} Searching for bulk load file: {} in paths: {}, {}", Utils.logPeerId(peerId),
+      relativePathFromNamespace, baseNamespaceDir, hFileArchiveDir);
+
     Path result =
       findExistingPath(rootFs, baseNamespaceDir, hFileArchiveDir, relativePathFromNamespace);
+
     if (result == null) {
+      LOG.error("{} No bulk loaded file found in relative path: {}", Utils.logPeerId(peerId),
+        relativePathFromNamespace);
       throw new IOException(
         "No Bulk loaded file found in relative path: " + relativePathFromNamespace);
     }
+
+    LOG.debug("{} Bulk load file found at {}", Utils.logPeerId(peerId), result);
     return result;
   }
 
   private static Path findExistingPath(FileSystem rootFs, Path baseNamespaceDir,
     Path hFileArchiveDir, Path filePath) throws IOException {
+    if (LOG.isTraceEnabled()) {
+      LOG.trace("Checking for bulk load file at: {} and {}", new Path(baseNamespaceDir, filePath),
+        new Path(hFileArchiveDir, filePath));
+    }
+
     for (Path candidate : new Path[] { new Path(baseNamespaceDir, filePath),
       new Path(hFileArchiveDir, filePath) }) {
       if (rootFs.exists(candidate)) {
+        LOG.debug("Found bulk load file at: {}", candidate);
         return candidate;
       }
     }
@@ -360,14 +418,16 @@ private static Path findExistingPath(FileSystem rootFs, Path baseNamespaceDir,
 
   private void shutdownFlushExecutor() {
     if (flushExecutor != null) {
+      LOG.info("{} Initiating WAL flush executor shutdown.", Utils.logPeerId(peerId));
+
       flushExecutor.shutdown();
       try {
         if (
           !flushExecutor.awaitTermination(EXECUTOR_TERMINATION_TIMEOUT_SECONDS, TimeUnit.SECONDS)
         ) {
-          flushExecutor.shutdownNow();
           LOG.warn("{} Flush executor did not terminate within timeout, forcing shutdown.",
             Utils.logPeerId(peerId));
+          flushExecutor.shutdownNow();
         }
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();

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

@@ -51,6 +51,10 @@ public void sync(boolean forceSync) throws IOException {
     if (fsDataOutputstream == null) {
       return; // Presume closed
     }
+    // Special case for Hadoop S3: Unlike traditional file systems, where flush() ensures data is
+    // durably written, in Hadoop S3, flush() only writes data to the internal buffer and does not
+    // immediately persist it to S3. The actual upload to S3 happens asynchronously, typically when
+    // a block is full or when close() is called, which finalizes the upload process.
     fsDataOutputstream.flush();
     AtomicUtils.updateMax(this.syncedLength, fsDataOutputstream.getPos());
   }