cleanup debug code

Author: zarna1parekh

astra/src/main/java/com/slack/astra/blobfs/BlobStore.java

@@ -100,13 +100,6 @@ public void uploadSequentially(String prefix, Path directoryToUpload) {
         String s3Key = prefix + "/" + relativePath;
 
         try {
-          LOG.info(
-              "Uploading file {}/{}: {} ({}MB)",
-              completedCount + 1,
-              allFiles.size(),
-              relativePath,
-              file.toFile().length() / (1024 * 1024));
-
           // Upload single file with checksum validation
           transferManager
               .uploadFile(
@@ -137,10 +130,11 @@ public void uploadSequentially(String prefix, Path directoryToUpload) {
       if (!failedUploads.isEmpty()) {
         throw new IllegalStateException(
             String.format(
-                "Some files failed to upload - attempted to upload %s files, failed %s.",
-                allFiles.size(), failedUploads.size()));
+                "Some files failed to upload for predix %s - attempted to upload %s files, failed %s.",
+                prefix, allFiles.size(), failedUploads.size()));
       }
-      LOG.info("Successfully uploaded all {} files sequentially", completedCount);
+      LOG.info(
+          "Successfully uploaded all {} files sequentially for predix {}", completedCount, prefix);
 
     } catch (IOException e) {
       throw new RuntimeException("Failed to walk directory", e);
@@ -206,48 +200,22 @@ public void download(String prefix, Path destinationDirectory) {
               .completionFuture()
               .get();
 
-      LOG.info(
-          "Downloaded directory from S3: bucket={}, prefix={}, destination={}, status={}",
-          bucketName,
-          prefix,
-          destinationDirectory,
-          download.failedTransfers());
-
       if (!download.failedTransfers().isEmpty()) {
         // Log each failed transfer with its exception
-        LOG.error(
-            "Error attempting to download directory from S3: bucket={}, prefix={}, destination={}, failed transfers={}",
-            bucketName,
-            prefix,
-            destinationDirectory,
-            download.failedTransfers());
         download
             .failedTransfers()
             .forEach(
                 failedFileDownload -> {
-                  Throwable cause = failedFileDownload.exception();
-                  while (cause.getCause() != null && cause != cause.getCause()) {
-                    cause = cause.getCause();
-                  }
-                  if (cause instanceof S3Exception s3ex) {
-                    LOG.error(
-                        "Error attempting to download file from S3: key={}, requestId={}, extendedRequestId={},",
-                        failedFileDownload.request().getObjectRequest().key(),
-                        s3ex.requestId(),
-                        s3ex.extendedRequestId(),
-                        s3ex);
-
-                  } else {
-                    LOG.error(
-                        "Error attempting to download file from S3",
-                        failedFileDownload.exception());
-                  }
+                  LOG.error(
+                      "Error attempting to download file from S3 for prefix {}",
+                      prefix,
+                      failedFileDownload.exception());
                 });
 
         throw new IllegalStateException(
             String.format(
-                "Some files failed to download - failed to download %s files.",
-                download.failedTransfers().size()));
+                "Some files failed to download for prefix %s - failed to download %s files.",
+                prefix, download.failedTransfers().size()));
       }
     } catch (ExecutionException | InterruptedException e) {
       throw new RuntimeException(e);

astra/src/main/java/com/slack/astra/blobfs/S3AsyncUtil.java

@@ -45,8 +45,7 @@ public static S3AsyncClient initS3Client(AstraConfigs.S3Config config) {
 
       // default to 5% of the heap size for the max crt off-heap or 1GiB (min for client)
       long jvmMaxHeapSizeBytes = Runtime.getRuntime().maxMemory();
-      long defaultCrtMemoryLimit =
-          Math.max(Math.round(jvmMaxHeapSizeBytes * 0.05), 4L * 1024 * 1024 * 1024);
+      long defaultCrtMemoryLimit = Math.max(Math.round(jvmMaxHeapSizeBytes * 0.05), 1073741824);
       long maxNativeMemoryLimitBytes =
           Long.parseLong(
               System.getProperty(
@@ -57,7 +56,7 @@ public static S3AsyncClient initS3Client(AstraConfigs.S3Config config) {
           maxNativeMemoryLimitBytes);
       S3CrtAsyncClientBuilder s3AsyncClient =
           S3AsyncClient.crtBuilder()
-              .retryConfiguration(S3CrtRetryConfiguration.builder().numRetries(10).build())
+              .retryConfiguration(S3CrtRetryConfiguration.builder().numRetries(3).build())
               .targetThroughputInGbps(config.getS3TargetThroughputGbps())
               .region(Region.of(region))
               .maxNativeMemoryLimitInBytes(maxNativeMemoryLimitBytes)

astra/src/main/java/com/slack/astra/chunk/ChunkValidationUtils.java

@@ -9,9 +9,9 @@ public class ChunkValidationUtils {
 
   public static boolean isChunkClean(Path path) throws Exception {
     FSDirectory existingDir = FSDirectory.open(path, NoLockFactory.INSTANCE);
-    CheckIndex checker = new CheckIndex(existingDir);
-    CheckIndex.Status status = checker.checkIndex();
-    checker.close();
-    return status.clean;
+    try (CheckIndex checker = new CheckIndex(existingDir)) {
+      CheckIndex.Status status = checker.checkIndex();
+      return status.clean;
+    }
   }
 }

astra/src/main/java/com/slack/astra/chunk/ReadOnlyChunkImpl.java

@@ -36,6 +36,7 @@
 import java.nio.file.Paths;
 import java.time.Instant;
 import java.util.EnumSet;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -231,13 +232,14 @@ public CacheNodeAssignment getCacheNodeAssignment() {
     return assignment;
   }
 
-  private boolean validateS3vsLocalDownLoad() throws Exception {
+  private boolean validateS3vsLocalDownLoad() {
     // check if the number of files in S3 matches the local directory
     Map<String, Long> filesWithSizeInS3 = blobStore.listFilesWithSize(snapshotMetadata.snapshotId);
 
-    Map<String, Long> localFiles;
+    Map<String, Long> localFilesSizeMap;
+    Map<String, Long> mismatchFilesSizeMap = new HashMap<String, Long>();
     try (Stream<Path> fileList = Files.list(dataDirectory)) {
-      localFiles =
+      localFilesSizeMap =
           fileList
               .filter(Files::isRegularFile)
               .collect(
@@ -249,11 +251,12 @@ private boolean validateS3vsLocalDownLoad() throws Exception {
       throw new RuntimeException(
           String.format("Error reading local files in directory %s", dataDirectory), e);
     }
-    if (localFiles.size() != filesWithSizeInS3.size()) {
+    if (localFilesSizeMap.size() != filesWithSizeInS3.size()) {
       LOG.error(
-          String.format(
-              "Mismatch in number of files in S3 (%s) and local directory (%s) for snapshot %s",
-              filesWithSizeInS3.size(), localFiles.size(), snapshotMetadata.toString()));
+          "Mismatch in number of files in S3 ({}) and local directory ({}) for snapshot {}",
+          filesWithSizeInS3.size(),
+          localFilesSizeMap.size(),
+          snapshotMetadata.toString());
       return false;
     }
 
@@ -262,18 +265,26 @@ private boolean validateS3vsLocalDownLoad() throws Exception {
       long s3Size = entry.getValue();
       String fileName = Paths.get(s3Path).getFileName().toString();
 
-      if (!localFiles.containsKey(fileName) || !localFiles.get(fileName).equals(s3Size)) {
-        LOG.error(
-            String.format(
-                "Mismatch for file %s in S3 and local directory of size %s for snapshot %s",
-                s3Path, s3Size, snapshotMetadata.toString()));
-        return false;
+      if (!localFilesSizeMap.containsKey(fileName)
+          || !localFilesSizeMap.get(fileName).equals(s3Size)) {
+        mismatchFilesSizeMap.put(fileName, s3Size);
       }
     }
-    LOG.info(
-        "No file mismatch found for snapshot id '{}' and local directory '{}'",
-        snapshotMetadata.snapshotId,
-        dataDirectory.toString());
+    if (!mismatchFilesSizeMap.isEmpty()) {
+      String mismatchFilesAndSize =
+          mismatchFilesSizeMap.entrySet().stream()
+              .map(
+                  e ->
+                      String.format(
+                          "%s (S3Size: %s LocalSize: %s)",
+                          e.getKey(), e.getValue(), localFilesSizeMap.get(e.getKey())))
+              .collect(Collectors.joining(", "));
+      LOG.error(
+          "Mismatch in file sizes between S3 and local directory for snapshot {}. Mismatch files: {}",
+          snapshotMetadata.toString(),
+          mismatchFilesAndSize);
+      return false;
+    }
     return true;
   }
 
@@ -320,7 +331,7 @@ public void downloadChunkData() {
               String.format(
                   "Mismatch in number or size of files in S3 and local directory for snapshot %s",
                   snapshotMetadata);
-          throw new IOException(errorString);
+          throw new RuntimeException(errorString);
         }
 
         // check if lucene index is valid and not corrupted
@@ -330,10 +341,6 @@ public void downloadChunkData() {
               String.format(
                   "Lucene index is not clean. Found issues for snapshot: %s.", snapshotMetadata));
         }
-        LOG.info(
-            "Lucene index is clean for snapshot id '{}' and local directory '{}'",
-            snapshotMetadata.snapshotId,
-            dataDirectory.toString());
 
         // check if schema file exists
         Path schemaPath = Path.of(dataDirectory.toString(), ReadWriteChunk.SCHEMA_FILE_NAME);
@@ -375,16 +382,7 @@ public void downloadChunkData() {
       // disregarding any errors
       setAssignmentState(
           getCacheNodeAssignment(), Metadata.CacheNodeAssignment.CacheNodeAssignmentState.EVICT);
-      LOG.error(
-          "Error handling chunk assignment for assignment: {}, snapshot id: {}, snapshot size: {}, replicaId: {}, replicaSet: {}, cacheNodeId: {}",
-          "Error handling chunk assignment for assignment: {}, snapshot id: {}, snapshot size: {}, replicaId: {}, replicaSet: {}, cacheNodeId: {}",
-          assignment.assignmentId,
-          assignment.snapshotId,
-          assignment.snapshotSize,
-          assignment.replicaId,
-          assignment.replicaSet,
-          assignment.cacheNodeId,
-          e);
+      LOG.error("Error handling chunk assignment", e);
       assignmentTimer.stop(chunkAssignmentTimerFailure);
     } finally {
       chunkAssignmentLock.unlock();