From 3db1e9636420e5f9988325994890d978aa8b5dc5 Mon Sep 17 00:00:00 2001 From: Gordon Smith Date: Fri, 22 Sep 2023 14:28:19 +0100 Subject: [PATCH 1/4] Split off 9.2.24 Signed-off-by: Gordon Smith --- commons-hpcc/pom.xml | 2 +- dfsclient/pom.xml | 2 +- pom.xml | 2 +- wsclient/pom.xml | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/commons-hpcc/pom.xml b/commons-hpcc/pom.xml index 2e967b536..73bf12efc 100644 --- a/commons-hpcc/pom.xml +++ b/commons-hpcc/pom.xml @@ -9,7 +9,7 @@ org.hpccsystems hpcc4j - 9.2.23-0-SNAPSHOT + 9.2.25-0-SNAPSHOT diff --git a/dfsclient/pom.xml b/dfsclient/pom.xml index 84ad0583a..9bbd92fa7 100644 --- a/dfsclient/pom.xml +++ b/dfsclient/pom.xml @@ -9,7 +9,7 @@ org.hpccsystems hpcc4j - 9.2.23-0-SNAPSHOT + 9.2.25-0-SNAPSHOT diff --git a/pom.xml b/pom.xml index 33772a474..dee5dbb69 100644 --- a/pom.xml +++ b/pom.xml @@ -4,7 +4,7 @@ 4.0.0 org.hpccsystems hpcc4j - 9.2.23-0-SNAPSHOT + 9.2.25-0-SNAPSHOT pom HPCC Systems Java Projects https://hpccsystems.com diff --git a/wsclient/pom.xml b/wsclient/pom.xml index bfb9a77f2..150baeedf 100644 --- a/wsclient/pom.xml +++ b/wsclient/pom.xml @@ -9,7 +9,7 @@ org.hpccsystems hpcc4j - 9.2.23-0-SNAPSHOT + 9.2.25-0-SNAPSHOT From 419feaf502874456c6fb7db6a1d6fe5b7d3a00c8 Mon Sep 17 00:00:00 2001 From: James McMullan Date: Tue, 26 Sep 2023 15:22:13 -0400 Subject: [PATCH 2/4] HPCC4J-543 Empty compressed file test failure (#645) - Introduced socket operation timeouts Signed-off-by: James McMullan James.McMullan@lexisnexis.com Signed-off-by: James McMullan James.McMullan@lexisnexis.com --- .../dfs/client/HPCCRemoteFileWriter.java | 28 ++++++++++++- .../dfs/client/HpccRemoteFileReader.java | 41 ++++++++++++++++--- .../dfs/client/RowServiceInputStream.java | 37 +++++++++++++++++ .../dfs/client/RowServiceOutputStream.java | 39 +++++++++++++++++- 4 files changed, 135 insertions(+), 10 deletions(-) diff --git a/dfsclient/src/main/java/org/hpccsystems/dfs/client/HPCCRemoteFileWriter.java b/dfsclient/src/main/java/org/hpccsystems/dfs/client/HPCCRemoteFileWriter.java index a55e077c9..5108c19ab 100644 --- a/dfsclient/src/main/java/org/hpccsystems/dfs/client/HPCCRemoteFileWriter.java +++ b/dfsclient/src/main/java/org/hpccsystems/dfs/client/HPCCRemoteFileWriter.java @@ -69,12 +69,36 @@ public HPCCRemoteFileWriter(DataPartition dp, FieldDef recordDef, IRecordAccesso * @param fileCompression * the file compression * @param connectTimeoutMs - * the socket timeout in ms (default is 1000) + * the socket timeout in ms (default is 5000) * @throws Exception * the exception */ public HPCCRemoteFileWriter(DataPartition dp, FieldDef recordDef, IRecordAccessor recordAccessor, CompressionAlgorithm fileCompression, int connectTimeoutMs) throws Exception + { + this(dp,recordDef,recordAccessor,fileCompression,RowServiceOutputStream.DEFAULT_CONNECT_TIMEOUT_MILIS, RowServiceOutputStream.DEFAULT_SOCKET_OP_TIMEOUT_MS); + } + + /** + * A remote file writer. + * + * @param dp + * the part of the file, name and location + * @param recordDef + * the record def + * @param recordAccessor + * the record accessor + * @param fileCompression + * the file compression + * @param connectTimeoutMs + * the socket timeout in ms (default is 5000) + * @param socketOpTimeoutMs + * the socket operation timeout in ms (default is 15000) + * @throws Exception + * the exception + */ + public HPCCRemoteFileWriter(DataPartition dp, FieldDef recordDef, IRecordAccessor recordAccessor, CompressionAlgorithm fileCompression, int connectTimeoutMs, int socketOpTimeoutMs) + throws Exception { this.recordDef = recordDef; this.dataPartition = dp; @@ -83,7 +107,7 @@ public HPCCRemoteFileWriter(DataPartition dp, FieldDef recordDef, IRecordAccesso this.outputStream = new RowServiceOutputStream(dataPartition.getCopyIP(0), dataPartition.getPort(), dataPartition.getUseSsl(), dataPartition.getFileAccessBlob(), this.recordDef, this.dataPartition.getThisPart(), this.dataPartition.getCopyPath(0), - fileCompression, connectTimeoutMs); + fileCompression, connectTimeoutMs, socketOpTimeoutMs); this.binaryRecordWriter = new BinaryRecordWriter(this.outputStream); this.binaryRecordWriter.initialize(this.recordAccessor); diff --git a/dfsclient/src/main/java/org/hpccsystems/dfs/client/HpccRemoteFileReader.java b/dfsclient/src/main/java/org/hpccsystems/dfs/client/HpccRemoteFileReader.java index 38e283215..78e3e6500 100644 --- a/dfsclient/src/main/java/org/hpccsystems/dfs/client/HpccRemoteFileReader.java +++ b/dfsclient/src/main/java/org/hpccsystems/dfs/client/HpccRemoteFileReader.java @@ -76,7 +76,7 @@ public HpccRemoteFileReader(DataPartition dp, FieldDef originalRD, IRecordBuilde * @param recBuilder * the IRecordBuilder used to construct records * @param connectTimeout - * the connection timeout in seconds, -1 for default + * the connection timeout in milliseconds, -1 for default * @throws Exception * the exception */ @@ -95,7 +95,7 @@ public HpccRemoteFileReader(DataPartition dp, FieldDef originalRD, IRecordBuilde * @param recBuilder * the IRecordBuilder used to construct records * @param connectTimeout - * the connection timeout in seconds, -1 for default + * the connection timeout in milliseconds, -1 for default * @param limit * the maximum number of records to read from the provided data partition, -1 specifies no limit * @throws Exception @@ -116,7 +116,7 @@ public HpccRemoteFileReader(DataPartition dp, FieldDef originalRD, IRecordBuilde * @param recBuilder * the IRecordBuilder used to construct records * @param connectTimeout - * the connection timeout in seconds, -1 for default + * the connection timeout in milliseconds, -1 for default * @param limit * the maximum number of records to read from the provided data partition, -1 specifies no limit * @param createPrefetchThread @@ -141,7 +141,7 @@ public HpccRemoteFileReader(DataPartition dp, FieldDef originalRD, IRecordBuilde * @param recBuilder * the IRecordBuilder used to construct records * @param connectTimeout - * the connection timeout in seconds, -1 for default + * the connection timeout in milliseconds, -1 for default * @param limit * the maximum number of records to read from the provided data partition, -1 specifies no limit * @param createPrefetchThread @@ -154,6 +154,35 @@ public HpccRemoteFileReader(DataPartition dp, FieldDef originalRD, IRecordBuilde * general exception */ public HpccRemoteFileReader(DataPartition dp, FieldDef originalRD, IRecordBuilder recBuilder, int connectTimeout, int limit, boolean createPrefetchThread, int readSizeKB, FileReadResumeInfo resumeInfo) throws Exception + { + this(dp, originalRD, recBuilder, connectTimeout, limit, true, readSizeKB, resumeInfo, RowServiceInputStream.DEFAULT_SOCKET_OP_TIMEOUT_MS); + } + + /** + * A remote file reader that reads the part identified by the HpccPart object using the record definition provided. + * + * @param dp + * the part of the file, name and location + * @param originalRD + * the record defintion for the dataset + * @param recBuilder + * the IRecordBuilder used to construct records + * @param connectTimeout + * the connection timeout in milliseconds, -1 for default + * @param limit + * the maximum number of records to read from the provided data partition, -1 specifies no limit + * @param createPrefetchThread + * the input stream should create and manage prefetching on its own thread. If false prefetch needs to be called on another thread periodically. + * @param readSizeKB + * read request size in KB, -1 specifies use default value + * @param resumeInfo + * FileReadeResumeInfo data required to restart a read from a particular point in a file, null for reading from start + * @param socketOpTimeoutMs + * Socket (read / write) operation timeout in milliseconds + * @throws Exception + * general exception + */ + public HpccRemoteFileReader(DataPartition dp, FieldDef originalRD, IRecordBuilder recBuilder, int connectTimeout, int limit, boolean createPrefetchThread, int readSizeKB, FileReadResumeInfo resumeInfo, int socketOpTimeoutMs) throws Exception { this.handlePrefetch = createPrefetchThread; this.originalRecordDef = originalRD; @@ -178,7 +207,7 @@ public HpccRemoteFileReader(DataPartition dp, FieldDef originalRD, IRecordBuilde if (resumeInfo == null) { - this.inputStream = new RowServiceInputStream(this.dataPartition, this.originalRecordDef, projectedRecordDefinition, connectTimeout, limit, createPrefetchThread, readSizeKB); + this.inputStream = new RowServiceInputStream(this.dataPartition, this.originalRecordDef, projectedRecordDefinition, connectTimeout, limit, createPrefetchThread, readSizeKB, null, false, socketOpTimeoutMs); this.binaryRecordReader = new BinaryRecordReader(this.inputStream); this.binaryRecordReader.initialize(this.recordBuilder); @@ -193,7 +222,7 @@ public HpccRemoteFileReader(DataPartition dp, FieldDef originalRD, IRecordBuilde restartInfo.streamPos = resumeInfo.inputStreamPos; restartInfo.tokenBin = resumeInfo.tokenBin; - this.inputStream = new RowServiceInputStream(this.dataPartition, this.originalRecordDef, projectedRecordDefinition, connectTimeout, limit, createPrefetchThread, readSizeKB, restartInfo); + this.inputStream = new RowServiceInputStream(this.dataPartition, this.originalRecordDef, projectedRecordDefinition, connectTimeout, limit, createPrefetchThread, readSizeKB, restartInfo, false, socketOpTimeoutMs); long bytesToSkip = resumeInfo.recordReaderStreamPos - resumeInfo.inputStreamPos; if (bytesToSkip < 0) { diff --git a/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceInputStream.java b/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceInputStream.java index 794c73b1c..795fd8478 100644 --- a/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceInputStream.java +++ b/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceInputStream.java @@ -108,7 +108,9 @@ public class RowServiceInputStream extends InputStream implements IProfilable private Socket sock; public static final int DEFAULT_CONNECT_TIMEOUT_MILIS = 5000; // 5 second connection timeout + public static final int DEFAULT_SOCKET_OP_TIMEOUT_MS = 15000; // 15 second timeout on read / write operations private int connectTimeout = DEFAULT_CONNECT_TIMEOUT_MILIS; + private int socketOpTimeoutMs = DEFAULT_SOCKET_OP_TIMEOUT_MS; private static final Charset HPCCCharSet = Charset.forName("ISO-8859-1"); private static final Logger log = LogManager.getLogger(RowServiceInputStream.class); @@ -293,6 +295,37 @@ public RowServiceInputStream(DataPartition dp, FieldDef rd, FieldDef pRd, int co * general exception */ public RowServiceInputStream(DataPartition dp, FieldDef rd, FieldDef pRd, int connectTimeout, int limit, boolean createPrefetchThread, int maxReadSizeInKB, RestartInformation restartInfo, boolean isFetching) throws Exception + { + this(dp, rd, pRd, connectTimeout, limit, createPrefetchThread, maxReadSizeInKB, restartInfo, isFetching, DEFAULT_SOCKET_OP_TIMEOUT_MS); + } + + /** + * A plain socket connect to a THOR node for remote read + * + * @param dp + * the data partition to read + * @param rd + * the JSON definition for the read input and output + * @param pRd + * the projected record definition + * @param connectTimeout + * the connection timeout in milliseconds + * @param limit + * the record limit to use for reading the dataset. -1 implies no limit + * @param createPrefetchThread + * Wether or not this inputstream should handle prefetching itself or if prefetch will be called externally + * @param maxReadSizeInKB + * max readsize in kilobytes + * @param restartInfo + * information used to restart a read from a particular stream position + * @param isFetching + * Will this input stream be used to serviced batched fetch requests + * @param socketOpTimeoutMS + * Socket (read / write) operation timeout in milliseconds + * @throws Exception + * general exception + */ + public RowServiceInputStream(DataPartition dp, FieldDef rd, FieldDef pRd, int connectTimeout, int limit, boolean createPrefetchThread, int maxReadSizeInKB, RestartInformation restartInfo, boolean isFetching, int socketOpTimeoutMS) throws Exception { this.recordDefinition = rd; this.projectedRecordDefinition = pRd; @@ -325,6 +358,7 @@ public RowServiceInputStream(DataPartition dp, FieldDef rd, FieldDef pRd, int co this.tokenBin = null; this.simulateFail = false; this.connectTimeout = connectTimeout; + this.socketOpTimeoutMs = socketOpTimeoutMS; this.recordLimit = limit; this.readBufferCapacity.set(this.maxReadSizeKB*1024*2); @@ -1530,6 +1564,9 @@ private void makeActive() throws HpccFileException sock.setPerformancePreferences(0, 1, 2); sock.connect(new InetSocketAddress(this.getIP(), this.dataPart.getPort()), this.connectTimeout); } + + this.sock.setSoTimeout(socketOpTimeoutMs); + log.debug("Connected: Remote address = " + sock.getInetAddress().toString() + " Remote port = " + sock.getPort()); } catch (java.net.UnknownHostException e) diff --git a/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceOutputStream.java b/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceOutputStream.java index 5e5b95ceb..f6ea288cd 100644 --- a/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceOutputStream.java +++ b/dfsclient/src/main/java/org/hpccsystems/dfs/client/RowServiceOutputStream.java @@ -40,6 +40,7 @@ public class RowServiceOutputStream extends OutputStream { private static final Logger log = LogManager.getLogger(RowServiceOutputStream.class); public static final int DEFAULT_CONNECT_TIMEOUT_MILIS = 5000; // 5 second connection timeout + public static final int DEFAULT_SOCKET_OP_TIMEOUT_MS = 15000; // 15 second timeout on reads private static int SCRATCH_BUFFER_LEN = 2048; private String rowServiceVersion = ""; @@ -50,6 +51,7 @@ public class RowServiceOutputStream extends OutputStream private int filePartIndex = -1; private String accessToken = null; private CompressionAlgorithm compressionAlgo = CompressionAlgorithm.NONE; + private int sockOpTimeoutMs = DEFAULT_SOCKET_OP_TIMEOUT_MS; private Socket socket = null; private DataInputStream dis = null; @@ -125,7 +127,6 @@ private static class RowServiceResponse this(ip,port,useSSL,accessToken,recordDef,filePartIndex,filePartPath,fileCompression, DEFAULT_CONNECT_TIMEOUT_MILIS); } - /** * Creates RowServiceOutputStream to be used to stream data to target dafilesrv on HPCC cluster. * @@ -146,12 +147,44 @@ private static class RowServiceResponse * @param fileCompression * the file compression * @param connectTimeoutMs - * the socket timeout in ms (default is 1000) + * the socket connect timeout in ms (default is 5000) * @throws Exception * the exception */ RowServiceOutputStream(String ip, int port, boolean useSSL, String accessToken, FieldDef recordDef, int filePartIndex, String filePartPath, CompressionAlgorithm fileCompression, int connectTimeoutMs) throws Exception + { + this(ip,port,useSSL,accessToken,recordDef,filePartIndex,filePartPath,fileCompression, connectTimeoutMs, DEFAULT_SOCKET_OP_TIMEOUT_MS); + } + + /** + * Creates RowServiceOutputStream to be used to stream data to target dafilesrv on HPCC cluster. + * + * @param ip + * the ip + * @param port + * the port + * @param useSSL + * the use SSL + * @param accessToken + * the access token + * @param recordDef + * the record def + * @param filePartIndex + * the file part index + * @param filePartPath + * the file part path + * @param fileCompression + * the file compression + * @param connectTimeoutMs + * the socket connect timeout in ms (default is 5000) + * @param socketOpTimeoutMS + * the socket operation(read/write) timeout in ms (default is 15000) + * @throws Exception + * the exception + */ + RowServiceOutputStream(String ip, int port, boolean useSSL, String accessToken, FieldDef recordDef, int filePartIndex, String filePartPath, + CompressionAlgorithm fileCompression, int connectTimeoutMs, int sockOpTimeoutMS) throws Exception { this.rowServiceIP = ip; this.rowServicePort = port; @@ -160,6 +193,7 @@ private static class RowServiceResponse this.filePath = filePartPath; this.accessToken = accessToken; this.compressionAlgo = fileCompression; + this.sockOpTimeoutMs = sockOpTimeoutMS; try { @@ -192,6 +226,7 @@ private static class RowServiceResponse this.socket.connect(new InetSocketAddress(rowServiceIP, rowServicePort), DEFAULT_CONNECT_TIMEOUT_MILIS); } + this.socket.setSoTimeout(sockOpTimeoutMs); this.dos = new DataOutputStream(socket.getOutputStream()); this.dis = new DataInputStream(socket.getInputStream()); From c7e303e5e7145e9275e7a7ea5ecb57180b22914d Mon Sep 17 00:00:00 2001 From: James McMullan Date: Thu, 28 Sep 2023 09:33:11 -0400 Subject: [PATCH 3/4] HPCC4J-532 Jirabot fixes (#646) - Manually set permissions on merge script Signed-off-by: James McMullan James.McMullan@lexisnexis.com Signed-off-by: James McMullan James.McMullan@lexisnexis.com --- .github/workflows/JirabotMerge.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/JirabotMerge.yml b/.github/workflows/JirabotMerge.yml index 8ee3c54ae..a35fb193d 100644 --- a/.github/workflows/JirabotMerge.yml +++ b/.github/workflows/JirabotMerge.yml @@ -7,6 +7,8 @@ on: - "master" - "candidate-*" +permissions: write-all + jobs: jirabot: runs-on: ubuntu-latest From 06aaa7bee848f07158eaa9d46598544ae79510b6 Mon Sep 17 00:00:00 2001 From: Rodrigo Date: Thu, 28 Sep 2023 10:01:15 -0400 Subject: [PATCH 4/4] HPCC4J-532 Extends print statement - Adds log statement info Signed-off-by: Rodrigo Pastrana Signed-off-by: Rodrigo --- .github/workflows/JirabotMerge.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/JirabotMerge.yml b/.github/workflows/JirabotMerge.yml index a35fb193d..3fc3dadf0 100644 --- a/.github/workflows/JirabotMerge.yml +++ b/.github/workflows/JirabotMerge.yml @@ -184,7 +184,7 @@ jobs: branch_name = os.environ['BRANCH_NAME'] comments_url = os.environ['COMMENTS_URL'] - print("%s %s %s" % (title, user, comments_url)) + print("Attempting to close out Jira issue: %s %s %s" % (title, user, comments_url)) result = '' issuem = re.search("(HPCC4J|JAPI)-[0-9]+", title) if issuem: