Merge remote-tracking branch 'origin/candidate-9.4.x' into candidate-9.6.x

jakesmith · jakesmith · commit 58e2f45c8bcd · 2024-05-03T16:40:35.000+01:00
Signed-off-by: Jake Smith &lt;jake.smith@lexisnexisrisk.com&gt;

# Conflicts:
#	commons-hpcc/pom.xml
#	dfsclient/pom.xml
#	pom.xml
#	wsclient/pom.xml
diff --git a/dfsclient/src/main/java/org/hpccsystems/dfs/client/FileUtility.java b/dfsclient/src/main/java/org/hpccsystems/dfs/client/FileUtility.java
@@ -64,6 +64,7 @@ public class FileUtility
     private static final int DEFAULT_SPLIT_TABLE_SIZE = 128;
 
     private static final int NUM_DEFAULT_THREADS = 4;
+    static private final int DEFAULT_ACCESS_EXPIRY_SECONDS = 120;
 
     private static class TaskContext
     {
@@ -425,6 +426,7 @@ private static Options getReadTestOptions()
         options.addOption("user", true, "Specifies the username used to connect. Defaults to null.");
         options.addOption("pass", true, "Specifies the password used to connect. Defaults to null.");
         options.addOption("num_threads", true, "Specifies the number of parallel to use to perform operations.");
+        options.addOption("access_expiry_seconds", true, "Access token expiration seconds.");
 
         options.addOption(Option.builder("file_parts")
                                 .argName("_file_parts")
@@ -633,11 +635,6 @@ private static String[] filterFilesByFormat(String[] srcFiles, FileFormat format
 
     private static void executeTasks(Runnable[] tasks, int numThreads) throws Exception
     {
-        if (tasks.length > numThreads)
-        {
-            numThreads = tasks.length;
-        }
-
         int numTasksPerThread = tasks.length / numThreads;
         int numResidualTasks = tasks.length % numThreads;
 
@@ -686,16 +683,15 @@ private static Runnable[] createReadTestTasks(DataPartition[] fileParts, FieldDe
         {
             final int taskIndex = i;
             final DataPartition filePart = fileParts[taskIndex];
-            final HpccRemoteFileReader<HPCCRecord> filePartReader = new HpccRemoteFileReader<HPCCRecord>(filePart, recordDef, new HPCCRecordBuilder(recordDef));
 
             tasks[taskIndex] = new Runnable()
             {
-                HpccRemoteFileReader<HPCCRecord> fileReader = filePartReader;
-
                 public void run()
                 {
                     try
                     {
+                        HpccRemoteFileReader<HPCCRecord> fileReader = new HpccRemoteFileReader<HPCCRecord>(filePart, recordDef, new HPCCRecordBuilder(recordDef));
+
                         while (fileReader.hasNext())
                         {
                             HPCCRecord record = fileReader.next();
@@ -1250,6 +1246,18 @@ private static void performReadTest(String[] args, TaskContext context)
                               + numThreadsStr + ", must be an integer. Defaulting to: " + NUM_DEFAULT_THREADS + " threads.");
         }
 
+        int expirySeconds = DEFAULT_ACCESS_EXPIRY_SECONDS;
+        String expirySecondsStr = cmd.getOptionValue("access_expiry_seconds", "" + expirySeconds);
+        try
+        {
+            expirySeconds = Integer.parseInt(expirySecondsStr);
+        }
+        catch(Exception e)
+        {
+            System.out.println("Invalid option value for access_expiry_seconds: "
+                              + numThreadsStr + ", must be an integer. Defaulting to: " + DEFAULT_ACCESS_EXPIRY_SECONDS + "s.");
+        }
+
         String formatStr = cmd.getOptionValue("format");
         if (formatStr == null)
         {
@@ -1277,6 +1285,7 @@ private static void performReadTest(String[] args, TaskContext context)
         try
         {
             file = new HPCCFile(datasetName, connString, user, pass);
+            file.setFileAccessExpirySecs(expirySeconds);
         }
         catch (Exception e)
         {
diff --git a/dfsclient/src/main/java/org/hpccsystems/dfs/client/HpccRemoteFileReader.java b/dfsclient/src/main/java/org/hpccsystems/dfs/client/HpccRemoteFileReader.java
@@ -38,12 +38,20 @@ public class HpccRemoteFileReader<T> implements Iterator<T>
     private boolean               handlePrefetch    = true;
     private boolean               isClosed          = false;
     private boolean               canReadNext       = true;
+    private boolean               createPrefetchThread = true;
+    private int                   retryCount        = 0;
+    private int                   connectTimeout    = 0;
+    private int                   readSizeKB        = 0;
+    private int                   limit             = -1;
+    private int                   maxReadRetries    = DEFAULT_READ_RETRIES;
+    private int                   socketOpTimeoutMs = 0;
     private long                  openTimeMs        = 0;
     private long                  recordsRead       = 0;
 
     public static final int    NO_RECORD_LIMIT                  = -1;
     public static final int    DEFAULT_READ_SIZE_OPTION         = -1;
     public static final int    DEFAULT_CONNECT_TIMEOUT_OPTION   = -1;
+    public static final int    DEFAULT_READ_RETRIES             = 3;
 
     public static class FileReadResumeInfo
     {
@@ -189,18 +197,23 @@ public HpccRemoteFileReader(DataPartition dp, FieldDef originalRD, IRecordBuilde
     {
         this.handlePrefetch = createPrefetchThread;
         this.originalRecordDef = originalRD;
-        if (this.originalRecordDef == null)
-        {
-            throw new Exception("HpccRemoteFileReader: Original record definition is null.");
-        }
+        this.dataPartition = dp;
+        this.recordBuilder = recBuilder;
+        this.readSizeKB = readSizeKB;
+        this.limit = limit;
+        this.createPrefetchThread = createPrefetchThread;
+        this.socketOpTimeoutMs = socketOpTimeoutMs;
 
         if (connectTimeout < 1)
         {
             connectTimeout = RowServiceInputStream.DEFAULT_CONNECT_TIMEOUT_MILIS;
         }
+        this.connectTimeout = connectTimeout;
 
-        this.dataPartition = dp;
-        this.recordBuilder = recBuilder;
+        if (this.originalRecordDef == null)
+        {
+            throw new Exception("HpccRemoteFileReader: Provided original record definition is null, original record definition is required.");
+        }
 
         FieldDef projectedRecordDefinition = recBuilder.getRecordDefinition();
         if (projectedRecordDefinition == null)
@@ -246,6 +259,61 @@ public HpccRemoteFileReader(DataPartition dp, FieldDef originalRD, IRecordBuilde
         openTimeMs = System.currentTimeMillis();
     }
 
+    private boolean retryRead()
+    {
+        if (retryCount < maxReadRetries)
+        {
+            log.info("Retrying read for " + this.dataPartition.toString() + " retry count: " + retryCount);
+            retryCount++;
+
+            FileReadResumeInfo resumeInfo = getFileReadResumeInfo();
+            RowServiceInputStream.RestartInformation restartInfo = new RowServiceInputStream.RestartInformation();
+            restartInfo.streamPos = resumeInfo.inputStreamPos;
+            restartInfo.tokenBin = resumeInfo.tokenBin;
+
+            try
+            {
+                this.inputStream.close();
+            }
+            catch (Exception e) {}
+
+            try
+            {
+                this.inputStream = new RowServiceInputStream(this.dataPartition, this.originalRecordDef,
+                                                            this.recordBuilder.getRecordDefinition(), this.connectTimeout, this.limit, this.createPrefetchThread,
+                                                            this.readSizeKB, restartInfo, false, this.socketOpTimeoutMs);
+                long bytesToSkip = resumeInfo.recordReaderStreamPos - resumeInfo.inputStreamPos;
+                if (bytesToSkip < 0)
+                {
+                    throw new Exception("Unable to restart read stream, unexpected stream position in record reader.");
+                }
+                this.inputStream.skip(bytesToSkip);
+
+                this.binaryRecordReader = new BinaryRecordReader(this.inputStream, resumeInfo.recordReaderStreamPos);
+                this.binaryRecordReader.initialize(this.recordBuilder);
+            }
+            catch (Exception e)
+            {
+                log.error("Failed to retry read for " + this.dataPartition.toString() + " " + e.getMessage(), e);
+                return false;
+            }
+
+            return true;
+        }
+
+        return false;
+    }
+
+    /**
+     * Sets the maximum number of times to retry a read operation before failing.
+     *
+     * @param maxReadRetries maximum number of read retries
+     */
+    public void setMaxReadRetries(int maxReadRetries)
+    {
+        this.maxReadRetries = maxReadRetries;
+    }
+
     /**
      * Returns the stream position within the file.
      *
@@ -363,11 +431,16 @@ public boolean hasNext()
         }
         catch (HpccFileException e)
         {
-            canReadNext = false;
-            log.error("Read failure for " + this.dataPartition.toString());
-            java.util.NoSuchElementException exception = new java.util.NoSuchElementException("Fatal read error: " + e.getMessage());
-            exception.initCause(e);
-            throw exception;
+            if (!retryRead())
+            {
+                canReadNext = false;
+                log.error("Read failure for " + this.dataPartition.toString(), e);
+                java.util.NoSuchElementException exception = new java.util.NoSuchElementException("Fatal read error: " + e.getMessage());
+                exception.initCause(e);
+                throw exception;
+            }
+
+            return hasNext();
         }
 
         return canReadNext;
@@ -393,10 +466,15 @@ public T next()
         }
         catch (HpccFileException e)
         {
-            log.error("Read failure for " + this.dataPartition.toString() + " " + e.getMessage());
-            java.util.NoSuchElementException exception = new java.util.NoSuchElementException("Fatal read error: " + e.getMessage());
-            exception.initCause(e);
-            throw exception;
+            if (!retryRead())
+            {
+                log.error("Read failure for " + this.dataPartition.toString() + " " + e.getMessage(), e);
+                java.util.NoSuchElementException exception = new java.util.NoSuchElementException("Fatal read error: " + e.getMessage());
+                exception.initCause(e);
+                throw exception;
+            }
+
+            return next();
         }
 
         recordsRead++;

Original file line number	Diff line number	Diff line change
`@@ -64,6 +64,7 @@ public class FileUtility`
`64`	`64`	`private static final int DEFAULT_SPLIT_TABLE_SIZE = 128;`
`65`	`65`
`66`	`66`	`private static final int NUM_DEFAULT_THREADS = 4;`
	`67`	`+ static private final int DEFAULT_ACCESS_EXPIRY_SECONDS = 120;`
`67`	`68`
`68`	`69`	`private static class TaskContext`
`69`	`70`	`{`
`@@ -425,6 +426,7 @@ private static Options getReadTestOptions()`
`425`	`426`	`options.addOption("user", true, "Specifies the username used to connect. Defaults to null.");`
`426`	`427`	`options.addOption("pass", true, "Specifies the password used to connect. Defaults to null.");`
`427`	`428`	`options.addOption("num_threads", true, "Specifies the number of parallel to use to perform operations.");`
	`429`	`+ options.addOption("access_expiry_seconds", true, "Access token expiration seconds.");`
`428`	`430`
`429`	`431`	`options.addOption(Option.builder("file_parts")`
`430`	`432`	`.argName("_file_parts")`
`@@ -633,11 +635,6 @@ private static String[] filterFilesByFormat(String[] srcFiles, FileFormat format`
`633`	`635`
`634`	`636`	`private static void executeTasks(Runnable[] tasks, int numThreads) throws Exception`
`635`	`637`	`{`
`636`		`- if (tasks.length > numThreads)`
`637`		`- {`
`638`		`- numThreads = tasks.length;`
`639`		`- }`
`640`		`-`
`641`	`638`	`int numTasksPerThread = tasks.length / numThreads;`
`642`	`639`	`int numResidualTasks = tasks.length % numThreads;`
`643`	`640`
`@@ -686,16 +683,15 @@ private static Runnable[] createReadTestTasks(DataPartition[] fileParts, FieldDe`
`686`	`683`	`{`
`687`	`684`	`final int taskIndex = i;`
`688`	`685`	`final DataPartition filePart = fileParts[taskIndex];`
`689`		`- final HpccRemoteFileReader<HPCCRecord> filePartReader = new HpccRemoteFileReader<HPCCRecord>(filePart, recordDef, new HPCCRecordBuilder(recordDef));`
`690`	`686`
`691`	`687`	`tasks[taskIndex] = new Runnable()`
`692`	`688`	`{`
`693`		`- HpccRemoteFileReader<HPCCRecord> fileReader = filePartReader;`
`694`		`-`
`695`	`689`	`public void run()`
`696`	`690`	`{`
`697`	`691`	`try`
`698`	`692`	`{`
	`693`	`+ HpccRemoteFileReader<HPCCRecord> fileReader = new HpccRemoteFileReader<HPCCRecord>(filePart, recordDef, new HPCCRecordBuilder(recordDef));`
	`694`	`+`
`699`	`695`	`while (fileReader.hasNext())`
`700`	`696`	`{`
`701`	`697`	`HPCCRecord record = fileReader.next();`
`@@ -1250,6 +1246,18 @@ private static void performReadTest(String[] args, TaskContext context)`
`1250`	`1246`	`+ numThreadsStr + ", must be an integer. Defaulting to: " + NUM_DEFAULT_THREADS + " threads.");`
`1251`	`1247`	`}`
`1252`	`1248`
	`1249`	`+ int expirySeconds = DEFAULT_ACCESS_EXPIRY_SECONDS;`
	`1250`	`+ String expirySecondsStr = cmd.getOptionValue("access_expiry_seconds", "" + expirySeconds);`
	`1251`	`+ try`
	`1252`	`+ {`
	`1253`	`+ expirySeconds = Integer.parseInt(expirySecondsStr);`
	`1254`	`+ }`
	`1255`	`+ catch(Exception e)`
	`1256`	`+ {`
	`1257`	`+ System.out.println("Invalid option value for access_expiry_seconds: "`
	`1258`	`+ + numThreadsStr + ", must be an integer. Defaulting to: " + DEFAULT_ACCESS_EXPIRY_SECONDS + "s.");`
	`1259`	`+ }`
	`1260`	`+`
`1253`	`1261`	`String formatStr = cmd.getOptionValue("format");`
`1254`	`1262`	`if (formatStr == null)`
`1255`	`1263`	`{`
`@@ -1277,6 +1285,7 @@ private static void performReadTest(String[] args, TaskContext context)`
`1277`	`1285`	`try`
`1278`	`1286`	`{`
`1279`	`1287`	`file = new HPCCFile(datasetName, connString, user, pass);`
	`1288`	`+ file.setFileAccessExpirySecs(expirySeconds);`
`1280`	`1289`	`}`
`1281`	`1290`	`catch (Exception e)`
`1282`	`1291`	`{`