Add logs for remote store metadata intermittent read failures (opense…

…arch-project#8618) --------- Signed-off-by: bansvaru <bansvaru@amazon.com> Signed-off-by: Shivansh Arora <hishiv@amazon.com>
Gaurav614 · Apr 25, 2024 · ad2867f · ad2867f
1 parent e8be7d5
commit ad2867f
Showing 1 changed file with 21 additions and 4 deletions.
diff --git a/server/src/main/java/org/opensearch/common/io/VersionedCodecStreamWrapper.java b/server/src/main/java/org/opensearch/common/io/VersionedCodecStreamWrapper.java
@@ -10,7 +10,11 @@
 
 import java.io.IOException;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.lucene.codecs.CodecUtil;
+import org.apache.lucene.index.CorruptIndexException;
 import org.apache.lucene.store.ChecksumIndexInput;
 import org.apache.lucene.store.IndexInput;
 import org.apache.lucene.store.IndexOutput;
@@ -22,6 +26,8 @@
  * @opensearch.internal
  */
 public class VersionedCodecStreamWrapper<T> {
+    private static final Logger logger = LogManager.getLogger(VersionedCodecStreamWrapper.class);
+
     // TODO This can be updated to hold a streamReadWriteHandlerFactory and get relevant handler based on the stream versions
     private final IndexIOStreamHandler<T> indexIOStreamHandler;
     private final int currentVersion;
@@ -46,10 +52,21 @@ public VersionedCodecStreamWrapper(IndexIOStreamHandler<T> indexIOStreamHandler,
      * @return stream content parsed into {@link T}
      */
     public T readStream(IndexInput indexInput) throws IOException {
-        CodecUtil.checksumEntireFile(indexInput);
-        int readStreamVersion = checkHeader(indexInput);
-        T content = getHandlerForVersion(readStreamVersion).readContent(indexInput);
-        return content;
+        logger.debug("Reading input stream [{}] of length - [{}]", indexInput.toString(), indexInput.length());
+        try {
+            CodecUtil.checksumEntireFile(indexInput);
+            int readStreamVersion = checkHeader(indexInput);
+            return getHandlerForVersion(readStreamVersion).readContent(indexInput);
+        } catch (CorruptIndexException cie) {
+            logger.error(
+                () -> new ParameterizedMessage(
+                    "Error while validating header/footer for [{}]. Total data length [{}]",
+                    indexInput.toString(),
+                    indexInput.length()
+                )
+            );
+            throw cie;
+        }
     }
 
     /**