This is an automated email from the ASF dual-hosted git repository.
mattcasters pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 2e60ed34cb issue #7507 : fix OpenSearch, MinIO, and samples
integration test failures (#7508)
2e60ed34cb is described below
commit 2e60ed34cb4d2689ac8f5abd8b020b9e6c66f478
Author: Matt Casters <[email protected]>
AuthorDate: Sun Jul 12 16:20:26 2026 +0200
issue #7507 : fix OpenSearch, MinIO, and samples integration test failures
(#7508)
Null-safe startDateFilter for OpenSearch/Neo4j execution info locations.
REST response JsonPath works with FORM URLENCODED requests (Slack sample).
Ignore Slack sample in CI without SLACK_BOT_TOKEN.
Rewrite MinIO VFS writes to buffer-then-upload with known size and fix
createFile/getOutputStream lifecycle so files are no longer empty.
Add CLIENT_UNZIP option to run-tests-docker.sh (default true) so a patched
target/hop can be reused with CLIENT_UNZIP=false.
Fix .dockerignore so the Hop client install is included in docker builds.
Local minio integration tests pass after these changes; confirm on Jenkins.
---
.dockerignore | 86 ++++++++----
.../samples/read-samples-build-hop-run.hpl | 4 +
integration-tests/scripts/run-tests-docker.sh | 23 ++-
.../org/apache/hop/vfs/minio/MinioFileObject.java | 25 ++--
.../org/apache/hop/vfs/minio/MinioFileSystem.java | 8 +-
.../hop/vfs/minio/util/MinioPipedOutputStream.java | 132 ++++++++----------
.../vfs/minio/util/MinioPipedOutputStreamTest.java | 155 ++++++++++++++-------
.../neo4j/execution/NeoExecutionInfoLocation.java | 5 +-
.../OpenSearchExecutionInfoLocation.java | 5 +-
.../apache/hop/pipeline/transforms/rest/Rest.java | 43 ++++--
.../transforms/rest/ResponseJsonParseTest.java | 82 +++++++++++
11 files changed, 383 insertions(+), 185 deletions(-)
diff --git a/.dockerignore b/.dockerignore
index 71f30a9ef7..ada15024a7 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -1,36 +1,62 @@
-# Ignore everything
+# Ignore everything by default, then re-include only what Dockerfiles need.
+# With a leading "*", every parent path of a re-included file must also be
re-included,
+# and directory contents need an explicit "/**" rule or they stay ignored.
*
-# Allow files and directories needed for OLD Dockerfiles (pre-built artifacts)
-!/assemblies/client/target/hop-*
-!/assemblies/client/target/hop
-!/integration-tests/scripts
-!/integration-tests/ssh/keys
-!/assemblies/web/target
-!/assemblies/plugins/target
-!/docker
-!/rest
+# ---- Pre-built Hop client used by unit-tests.Dockerfile ----
+!assemblies/
+!assemblies/client/
+!assemblies/client/target/
+!assemblies/client/target/hop/
+!assemblies/client/target/hop/**
+!assemblies/client/target/hop-*
+
+# ---- Integration-test resources ----
+!integration-tests/
+!integration-tests/scripts/
+!integration-tests/scripts/**
+!integration-tests/ssh/
+!integration-tests/ssh/keys/
+!integration-tests/ssh/keys/**
+
+# ---- Other assembly artefacts ----
+!assemblies/web/
+!assemblies/web/target/
+!assemblies/web/target/**
+!assemblies/plugins/
+!assemblies/plugins/target/
+!assemblies/plugins/target/**
+
+# ---- Docker build context ----
+!docker/
+!docker/**
+
+# ---- REST / misc ----
+!rest/
+!rest/**
!google-key-apache-hop-it.json
-# Allow source files needed for NEW unified Dockerfile (maven build in Docker)
+# ---- Source tree for unified Dockerfile (maven build inside Docker) ----
!pom.xml
-!/*/pom.xml
-!/*/*/pom.xml
-!/*/*/*/pom.xml
-!/src
-!/*/src
-!/*/*/src
-!/*/*/*/src
-!/core
-!/engine
-!/engine-beam
-!/plugins
-!/assemblies
-!/ui
-!/rap
-!/rcp
-!/lib
-!/lib-jdbc
-!/.mvn
!mvnw
-!mvnw.cmd
\ No newline at end of file
+!mvnw.cmd
+!.mvn/
+!.mvn/**
+!core/
+!core/**
+!engine/
+!engine/**
+!engine-beam/
+!engine-beam/**
+!plugins/
+!plugins/**
+!ui/
+!ui/**
+!rap/
+!rap/**
+!rcp/
+!rcp/**
+!lib/
+!lib/**
+!lib-jdbc/
+!lib-jdbc/**
diff --git a/integration-tests/samples/read-samples-build-hop-run.hpl
b/integration-tests/samples/read-samples-build-hop-run.hpl
index 03643068f5..4397cc9e31 100644
--- a/integration-tests/samples/read-samples-build-hop-run.hpl
+++ b/integration-tests/samples/read-samples-build-hop-run.hpl
@@ -555,6 +555,10 @@ var hop_run_cmd = './hop-run.sh -j samples -r ' +
run_config + ' -f ' + filename
<item>table-compare.hpl</item>
<item>requires connection</item>
</line>
+ <line>
+ <item>rest-client-slack-conversations-loop.hpl</item>
+ <item>requires SLACK_BOT_TOKEN with channels:read or
conversations:read</item>
+ </line>
</data>
<distribute>Y</distribute>
<copies>1</copies>
diff --git a/integration-tests/scripts/run-tests-docker.sh
b/integration-tests/scripts/run-tests-docker.sh
index 446a91e285..07787d9630 100755
--- a/integration-tests/scripts/run-tests-docker.sh
+++ b/integration-tests/scripts/run-tests-docker.sh
@@ -35,6 +35,7 @@ for ARGUMENT in "$@"; do
JENKINS_GID) JENKINS_GID=${VALUE} ;;
GCP_KEY_FILE) GCP_KEY_FILE=${VALUE} ;;
KEEP_IMAGES) KEEP_IMAGES=${VALUE} ;;
+ CLIENT_UNZIP) CLIENT_UNZIP=${VALUE} ;;
*) ;;
esac
@@ -76,13 +77,31 @@ if [ -z "${KEEP_IMAGES}" ]; then
KEEP_IMAGES="false"
fi
+# Unzip the client zip into assemblies/client/target/hop by default. Set
CLIENT_UNZIP=false to
+# skip when target/hop already exists (e.g. after patching a single plugin
jar).
+if [ -z "${CLIENT_UNZIP}" ]; then
+ CLIENT_UNZIP="true"
+fi
+
# Cleanup surefire reports
rm -rf "${CURRENT_DIR}"/../surefire-reports
mkdir -p "${CURRENT_DIR}"/../surefire-reports/
chmod 777 "${CURRENT_DIR}"/../surefire-reports/
-# Unzip Hop
-unzip -o -q "${CURRENT_DIR}/../../assemblies/client/target/*.zip" -d
${CURRENT_DIR}/../../assemblies/client/target/
+HOP_CLIENT_TARGET_DIR="${CURRENT_DIR}/../../assemblies/client/target"
+HOP_DIR="${HOP_CLIENT_TARGET_DIR}/hop"
+
+# Unzip Hop client unless skipped and a usable hop folder is already present
+if [ "${CLIENT_UNZIP}" = "true" ] || [ ! -d "${HOP_DIR}" ]; then
+ if [ "${CLIENT_UNZIP}" != "true" ] && [ ! -d "${HOP_DIR}" ]; then
+ echo "CLIENT_UNZIP=${CLIENT_UNZIP} but ${HOP_DIR} does not exist;
unzipping client zip"
+ else
+ echo "Unzipping Hop client into ${HOP_CLIENT_TARGET_DIR}
(CLIENT_UNZIP=${CLIENT_UNZIP})"
+ fi
+ unzip -o -q "${HOP_CLIENT_TARGET_DIR}"/*.zip -d "${HOP_CLIENT_TARGET_DIR}/"
+else
+ echo "Skipping client unzip (CLIENT_UNZIP=${CLIENT_UNZIP}, using existing
${HOP_DIR})"
+fi
# Build base image only once
docker compose -f ${DOCKER_FILES_DIR}/integration-tests-base.yaml build
--build-arg JENKINS_USER=${JENKINS_USER} --build-arg JENKINS_UID=${JENKINS_UID}
--build-arg JENKINS_GROUP=${JENKINS_GROUP} --build-arg
JENKINS_GID=${JENKINS_GID} --build-arg GCP_KEY_FILE=${GCP_KEY_FILE}
diff --git
a/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/MinioFileObject.java
b/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/MinioFileObject.java
index 2496a728b6..df806d1a90 100644
---
a/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/MinioFileObject.java
+++
b/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/MinioFileObject.java
@@ -370,12 +370,20 @@ public class MinioFileObject extends
AbstractFileObject<MinioFileSystem> {
@Override
public void createFile() throws FileSystemException {
try {
- OutputStream os = doGetOutputStream(false);
-
- // Open and close it to create an empty file.
- os.close();
+ // Create an empty object directly. Do not go through
doGetOutputStream(): HopVfs always
+ // calls createFile() then getOutputStream(), and reusing a stream that
createFile already
+ // closed left empty objects and "already closed" errors on the real
write path.
+ PutObjectArgs args =
+ PutObjectArgs.builder()
+ .bucket(bucketName)
+ .object(key)
+ .contentType("application/octet-stream")
+ .stream(new ByteArrayInputStream(new byte[0]), 0, -1)
+ .build();
+ fileSystem.getClient().putObject(args);
+ fileSystem.invalidateListCacheForParentOf(bucketName, key);
+ // Force re-attach so subsequent size/type reads are not stuck on
IMAGINARY.
closeMinio();
- // Update the metadata
doAttach();
} catch (Exception e) {
throw new FileSystemException("error creating file", e);
@@ -384,11 +392,8 @@ public class MinioFileObject extends
AbstractFileObject<MinioFileSystem> {
@Override
public OutputStream doGetOutputStream(boolean bAppend) throws Exception {
- if (outputStream != null) {
- // createFile() already built the output stream.
- return outputStream;
- }
-
+ // Always hand out a fresh stream. Do not close any previous stream here:
HopVfs may still be
+ // wrapping it, and closing early left empty objects after
createFile()+getOutputStream().
outputStream = new MinioPipedOutputStream(fileSystem, bucketName, key);
return outputStream;
}
diff --git
a/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/MinioFileSystem.java
b/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/MinioFileSystem.java
index 26d33b12bd..a96999daef 100644
---
a/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/MinioFileSystem.java
+++
b/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/MinioFileSystem.java
@@ -140,21 +140,21 @@ public class MinioFileSystem extends AbstractFileSystem {
this.listCacheTtlMs = listCacheTtlMs;
}
- void putListCache(
+ public void putListCache(
String bucket, String prefix, java.util.Map<String,
MinioListCache.ChildInfo> entries) {
getListCache().put(bucket, prefix, entries);
}
- MinioListCache.ChildInfo getFromListCache(
+ public MinioListCache.ChildInfo getFromListCache(
String bucket, String parentPrefix, String childFullKey) {
return getListCache().get(bucket, parentPrefix, childFullKey);
}
- void invalidateListCache(String bucket, String prefix) {
+ public void invalidateListCache(String bucket, String prefix) {
getListCache().invalidate(bucket, prefix);
}
- void invalidateListCacheForParentOf(String bucket, String key) {
+ public void invalidateListCacheForParentOf(String bucket, String key) {
getListCache().invalidateParentOf(bucket, key);
}
}
diff --git
a/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/util/MinioPipedOutputStream.java
b/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/util/MinioPipedOutputStream.java
index 5666669c65..44a65351d0 100644
---
a/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/util/MinioPipedOutputStream.java
+++
b/plugins/tech/minio/src/main/java/org/apache/hop/vfs/minio/util/MinioPipedOutputStream.java
@@ -18,112 +18,96 @@
package org.apache.hop.vfs.minio.util;
import io.minio.PutObjectArgs;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
-import java.io.InterruptedIOException;
-import java.io.PipedInputStream;
-import java.io.PipedOutputStream;
+import java.io.OutputStream;
import lombok.Getter;
import lombok.Setter;
import org.apache.hop.vfs.minio.MinioFileSystem;
-/** Custom OutputStream that enables an output stream onto Minio */
+/**
+ * OutputStream that buffers content and uploads it to MinIO on {@link
#close()}.
+ *
+ * <p>Previously this used a {@link java.io.PipedOutputStream} with the MinIO
SDK's unknown-size
+ * ({@code objectSize = -1}) multipart path. That combination races with
VFS/Text File Output
+ * close/flush chains and produced empty objects ({@code Pipe closed} on
write). Buffering and
+ * uploading with a known content length matches the reliable path used for
folder markers.
+ *
+ * <p>Close is idempotent and post-close flush is a no-op so nested VFS/Text
File Output close
+ * chains (which flush after closing the inner stream) do not fail the
pipeline.
+ */
@Getter
@Setter
-public class MinioPipedOutputStream extends PipedOutputStream {
+public class MinioPipedOutputStream extends OutputStream {
- private boolean initialized = false;
- private boolean blockedUntilDone = true;
- private final PipedInputStream pipedInputStream;
- private final MinioAsyncTransferRunner asyncTransferRunner;
+ private final ByteArrayOutputStream buffer = new ByteArrayOutputStream(64 *
1024);
private final MinioFileSystem fileSystem;
-
- // Daemon-threaded so a wedged putObject (black-holed socket) can never pin
the JVM, and the
- // uploader is referenced so close() can join() it interruptibly instead of
polling forever.
- private Thread transferThread;
- private volatile Exception transferError;
-
private final String bucketName;
private final String key;
- public MinioPipedOutputStream(MinioFileSystem fileSystem, String bucketName,
String key)
- throws IOException {
- this.pipedInputStream = new PipedInputStream(100000);
- try {
- this.pipedInputStream.connect(this);
- } catch (IOException e) {
- throw new IOException("A MinIO piped output stream could not connect to
the input stream", e);
- }
+ private boolean closed;
+ private boolean blockedUntilDone = true;
- this.asyncTransferRunner = new MinioAsyncTransferRunner();
+ public MinioPipedOutputStream(MinioFileSystem fileSystem, String bucketName,
String key) {
+ this.fileSystem = fileSystem;
this.bucketName = bucketName;
this.key = key;
- this.fileSystem = fileSystem;
- }
-
- private void initializeWrite() {
- if (!initialized) {
- initialized = true;
- transferThread = new Thread(asyncTransferRunner, "minio-upload-" +
bucketName + "-" + key);
- transferThread.setDaemon(true);
- transferThread.start();
- }
}
@Override
public void write(int b) throws IOException {
- initializeWrite();
- super.write(b);
+ // Outer VFS/Text File Output close chains may flush after the first
close. Accept late bytes
+ // and re-open so a subsequent close() can upload the complete content.
+ if (closed) {
+ closed = false;
+ }
+ buffer.write(b);
}
@Override
public void write(byte[] b, int off, int len) throws IOException {
- initializeWrite();
- super.write(b, off, len);
+ if (len == 0) {
+ return;
+ }
+ if (closed) {
+ closed = false;
+ }
+ buffer.write(b, off, len);
}
@Override
- public synchronized void flush() {
- // no flush
+ public void flush() {
+ // Data is held in memory until close(); nothing to flush to the network.
+ // No-op when closed so VFS MonitorOutputStream / FilterOutputStream
double-close is safe.
}
@Override
public void close() throws IOException {
- // Start upload even when no bytes were written (0-byte file), so
PutObject is still executed
- initializeWrite();
- super.close();
- if (initialized && isBlockedUntilDone()) {
- try {
- // join() is interruptible: a pipeline stop interrupts the transform
thread and now
- // breaks out of here instead of looping forever on a stalled upload.
- transferThread.join();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- transferThread.interrupt();
- throw new InterruptedIOException(
- "Interrupted while finishing the MinIO upload for " + bucketName +
"/" + key);
- }
- if (transferError != null) {
- throw new IOException("MinIO upload failed for " + bucketName + "/" +
key, transferError);
- }
+ if (closed) {
+ return;
}
- }
+ closed = true;
- class MinioAsyncTransferRunner implements Runnable {
- @Override
- public void run() {
- try {
- PutObjectArgs args =
- PutObjectArgs.builder()
- .contentType("application/octet-stream")
- .bucket(bucketName)
- .object(key)
- .stream(pipedInputStream, -1, fileSystem.getPartSize())
- .build();
- fileSystem.getClient().putObject(args);
- } catch (Exception e) {
- // Recorded and re-thrown from close() so the failure is surfaced
instead of swallowed.
- transferError = e;
- }
+ byte[] data = buffer.toByteArray();
+ try {
+ PutObjectArgs args =
+ PutObjectArgs.builder()
+ .contentType("application/octet-stream")
+ .bucket(bucketName)
+ .object(key)
+ .stream(new ByteArrayInputStream(data), data.length, -1)
+ .build();
+ fileSystem.getClient().putObject(args);
+ fileSystem.invalidateListCacheForParentOf(bucketName, key);
+ } catch (Exception e) {
+ throw new IOException("MinIO upload failed for " + bucketName + "/" +
key, e);
+ } finally {
+ buffer.reset();
}
}
+
+ public boolean isClosed() {
+ return closed;
+ }
}
diff --git
a/plugins/tech/minio/src/test/java/org/apache/hop/vfs/minio/util/MinioPipedOutputStreamTest.java
b/plugins/tech/minio/src/test/java/org/apache/hop/vfs/minio/util/MinioPipedOutputStreamTest.java
index bba47ae1e3..5ff1daf4cc 100644
---
a/plugins/tech/minio/src/test/java/org/apache/hop/vfs/minio/util/MinioPipedOutputStreamTest.java
+++
b/plugins/tech/minio/src/test/java/org/apache/hop/vfs/minio/util/MinioPipedOutputStreamTest.java
@@ -17,76 +17,72 @@
package org.apache.hop.vfs.minio.util;
-import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import io.minio.MinioClient;
import io.minio.PutObjectArgs;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import java.io.InputStream;
import org.apache.hop.vfs.minio.MinioFileSystem;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
/**
- * Regression guard for the MinIO VFS write-path forever-hang — same fix shape
as {@code
- * S3CommonPipedOutputStream}: {@link MinioPipedOutputStream#close()} now
joins the uploader thread
- * interruptibly and surfaces upload failures instead of polling forever and
swallowing them.
+ * Regression guard for MinIO VFS writes: content must be uploaded with a
known length (no
+ * pipe+unknown-size races) and upload failures must surface from {@code
close()}.
*/
class MinioPipedOutputStreamTest {
@Test
- void closeIsInterruptibleWhenPutObjectStalls() throws Exception {
- CountDownLatch putEntered = new CountDownLatch(1);
- CountDownLatch release = new CountDownLatch(1); // models a stalled MinIO
endpoint
+ void closeUploadsBufferedContentWithKnownLength() throws Exception {
+ MinioClient client = mock(MinioClient.class);
+ when(client.putObject(any(PutObjectArgs.class))).thenReturn(null);
+
+ MinioFileSystem fileSystem = mock(MinioFileSystem.class);
+ when(fileSystem.getClient()).thenReturn(client);
+
+ MinioPipedOutputStream out =
+ new MinioPipedOutputStream(fileSystem, "test-bucket", "path/file.txt");
+ byte[] payload = "hello-minio".getBytes();
+ out.write(payload);
+ out.close();
+
+ ArgumentCaptor<PutObjectArgs> argsCaptor =
ArgumentCaptor.forClass(PutObjectArgs.class);
+ verify(client).putObject(argsCaptor.capture());
+ PutObjectArgs args = argsCaptor.getValue();
+ assertEquals("test-bucket", args.bucket());
+ assertEquals("path/file.txt", args.object());
+ assertEquals(payload.length, args.objectSize());
+
+ try (InputStream stream = args.stream()) {
+ assertArrayEquals(payload, stream.readAllBytes());
+ }
+ verify(fileSystem).invalidateListCacheForParentOf(eq("test-bucket"),
eq("path/file.txt"));
+ }
+
+ @Test
+ void closeWithNoWritesUploadsEmptyObject() throws Exception {
MinioClient client = mock(MinioClient.class);
- when(client.putObject(any(PutObjectArgs.class)))
- .thenAnswer(
- inv -> {
- putEntered.countDown();
- release.await();
- return null;
- });
+ when(client.putObject(any(PutObjectArgs.class))).thenReturn(null);
MinioFileSystem fileSystem = mock(MinioFileSystem.class);
when(fileSystem.getClient()).thenReturn(client);
- when(fileSystem.getPartSize()).thenReturn(5L * 1024 * 1024);
- MinioPipedOutputStream out = new MinioPipedOutputStream(fileSystem,
"test-bucket", "test-key");
+ MinioPipedOutputStream out = new MinioPipedOutputStream(fileSystem,
"bucket", "empty.txt");
+ out.close();
- Thread closer =
- new Thread(
- () -> {
- try {
- out.close();
- } catch (Exception ignored) {
- // InterruptedIOException expected once we interrupt
- }
- },
- "minio-piped-closer");
- closer.setDaemon(true);
- closer.start();
-
- try {
- assertTrue(putEntered.await(5, TimeUnit.SECONDS), "background putObject
should have started");
-
- closer.join(1500);
- assertTrue(closer.isAlive(), "close() should still be waiting for the
in-progress upload");
-
- closer.interrupt();
- closer.join(3000);
- assertFalse(
- closer.isAlive(),
- "close() did not respond to interruption — the uninterruptible
forever-hang has been "
- + "reintroduced");
- } finally {
- release.countDown();
- }
+ ArgumentCaptor<PutObjectArgs> argsCaptor =
ArgumentCaptor.forClass(PutObjectArgs.class);
+ verify(client).putObject(argsCaptor.capture());
+ assertEquals(0L, argsCaptor.getValue().objectSize());
}
@Test
@@ -97,7 +93,6 @@ class MinioPipedOutputStreamTest {
MinioFileSystem fileSystem = mock(MinioFileSystem.class);
when(fileSystem.getClient()).thenReturn(client);
- when(fileSystem.getPartSize()).thenReturn(5L * 1024 * 1024);
MinioPipedOutputStream out = new MinioPipedOutputStream(fileSystem,
"test-bucket", "test-key");
out.write(new byte[] {1, 2, 3});
@@ -107,4 +102,68 @@ class MinioPipedOutputStreamTest {
out::close,
"a failed MinIO upload must be reported from close(), not silently
swallowed");
}
+
+ @Test
+ void writeAfterCloseIsAcceptedAndUploadedOnNextClose() throws Exception {
+ // Nested VFS/Text File Output close chains may flush after the first
close; late bytes must
+ // still be uploaded on a subsequent close().
+ MinioClient client = mock(MinioClient.class);
+ when(client.putObject(any(PutObjectArgs.class))).thenReturn(null);
+
+ MinioFileSystem fileSystem = mock(MinioFileSystem.class);
+ when(fileSystem.getClient()).thenReturn(client);
+
+ MinioPipedOutputStream out = new MinioPipedOutputStream(fileSystem,
"test-bucket", "key");
+ out.close(); // first close uploads empty
+
+ out.write(new byte[] {1, 2, 3});
+ out.close(); // second close uploads the late bytes
+
+ // first empty + second with content
+ verify(client,
org.mockito.Mockito.times(2)).putObject(any(PutObjectArgs.class));
+ }
+
+ @Test
+ void closeIsIdempotent() throws Exception {
+ MinioClient client = mock(MinioClient.class);
+ when(client.putObject(any(PutObjectArgs.class))).thenReturn(null);
+
+ MinioFileSystem fileSystem = mock(MinioFileSystem.class);
+ when(fileSystem.getClient()).thenReturn(client);
+
+ MinioPipedOutputStream out = new MinioPipedOutputStream(fileSystem,
"test-bucket", "key");
+ out.write(new byte[] {9});
+ out.close();
+ out.close(); // second close must not re-upload or throw
+
+ verify(client).putObject(any(PutObjectArgs.class));
+ }
+
+ @Test
+ void largeWriteIsFullyUploaded() throws Exception {
+ MinioClient client = mock(MinioClient.class);
+ when(client.putObject(any(PutObjectArgs.class))).thenReturn(null);
+
+ MinioFileSystem fileSystem = mock(MinioFileSystem.class);
+ when(fileSystem.getClient()).thenReturn(client);
+
+ byte[] payload = new byte[256 * 1024];
+ for (int i = 0; i < payload.length; i++) {
+ payload[i] = (byte) (i % 251);
+ }
+
+ MinioPipedOutputStream out = new MinioPipedOutputStream(fileSystem,
"test-bucket", "large.bin");
+ out.write(payload);
+ out.close();
+
+ ArgumentCaptor<PutObjectArgs> argsCaptor =
ArgumentCaptor.forClass(PutObjectArgs.class);
+ verify(client).putObject(argsCaptor.capture());
+ assertEquals(payload.length, argsCaptor.getValue().objectSize());
+
+ try (InputStream stream = argsCaptor.getValue().stream();
+ ByteArrayOutputStream readBack = new ByteArrayOutputStream()) {
+ stream.transferTo(readBack);
+ assertArrayEquals(payload, readBack.toByteArray());
+ }
+ }
}
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
index 1ea5ea7728..b0445b34eb 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
@@ -710,9 +710,10 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
}
builder.withExtraClause("n." + EP_EXECUTION_TYPE + " IN [ 'Workflow',
'Pipeline' ]");
}
- if (selector.startDateFilter() != LastPeriod.NONE) {
+ LastPeriod dateFilter = selector.startDateFilter();
+ if (dateFilter != null && dateFilter != LastPeriod.NONE) {
builder.withExtraClause(" AND n." + EP_EXECUTION_START_DATE + " >=
$fromStartDate ");
- builder.parameters().put("fromStartDate",
selector.startDateFilter().calculateStartDate());
+ builder.parameters().put("fromStartDate",
dateFilter.calculateStartDate());
}
}
diff --git
a/plugins/tech/opensearch/src/main/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocation.java
b/plugins/tech/opensearch/src/main/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocation.java
index f23228fefe..994d394a3e 100644
---
a/plugins/tech/opensearch/src/main/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocation.java
+++
b/plugins/tech/opensearch/src/main/java/org/apache/hop/execution/opensearch/OpenSearchExecutionInfoLocation.java
@@ -430,11 +430,12 @@ public class OpenSearchExecutionInfoLocation extends
BaseCachingExecutionInfoLoc
body = body.replace("__FROM_CLAUSE__", "FROM " + actualIndexName);
String whereClause = "";
- if (selector != null && selector.startDateFilter() != LastPeriod.NONE) {
+ LastPeriod dateFilter = selector != null ? selector.startDateFilter() :
null;
+ if (dateFilter != null && dateFilter != LastPeriod.NONE) {
// OpenSearch uses UTC and the GUI runs in local time.
//
ZonedDateTime localStartDate =
-
selector.startDateFilter().calculateStartDate().atZone(ZoneId.systemDefault());
+ dateFilter.calculateStartDate().atZone(ZoneId.systemDefault());
Date localStart = new Date(localStartDate.toInstant().toEpochMilli());
SimpleDateFormat whereFormat = new SimpleDateFormat("yyyy-MM-dd
HH:mm:ss");
diff --git
a/plugins/transforms/rest/src/main/java/org/apache/hop/pipeline/transforms/rest/Rest.java
b/plugins/transforms/rest/src/main/java/org/apache/hop/pipeline/transforms/rest/Rest.java
index f4d83347a8..6a761bd818 100644
---
a/plugins/transforms/rest/src/main/java/org/apache/hop/pipeline/transforms/rest/Rest.java
+++
b/plugins/transforms/rest/src/main/java/org/apache/hop/pipeline/transforms/rest/Rest.java
@@ -865,7 +865,9 @@ public class Rest extends BaseTransform<RestMeta, RestData>
{
if ("[]".equals(bodyTrim)) {
return true;
}
- if
(!RestMeta.APPLICATION_TYPE_JSON.equals(NVL(resolve(meta.getApplicationType()),
""))) {
+ // Response may be JSON even when the request Content-Type is
form-urlencoded (e.g. Slack).
+ if (!shouldParseResponseAsJson(
+ NVL(resolve(meta.getApplicationType()), ""), resolvedSplitPath)) {
return false;
}
try {
@@ -886,6 +888,19 @@ public class Rest extends BaseTransform<RestMeta,
RestData> {
}
}
+ /**
+ * Whether a response body should be parsed as JSON for pagination/split.
Application type is the
+ * <em>request</em> Content-Type; APIs such as Slack POST form bodies and
still return JSON.
+ * JsonPath expressions (starting with {@code $}) imply a JSON response
regardless of request
+ * type.
+ */
+ static boolean shouldParseResponseAsJson(String applicationType, String
pathOrExpression) {
+ if (RestMeta.APPLICATION_TYPE_JSON.equals(NVL(applicationType, ""))) {
+ return true;
+ }
+ return pathOrExpression != null && pathOrExpression.trim().startsWith("$");
+ }
+
private int emitPagedResultRows(
Object[] rowTemplate, String resolvedSplitPath, RestExchangeResult ex)
throws HopException {
@@ -916,7 +931,9 @@ public class Rest extends BaseTransform<RestMeta, RestData>
{
List<String> out = new ArrayList<>();
try {
- if (RestMeta.APPLICATION_TYPE_JSON.equals(appType)) {
+ // Prefer JsonPath when the expression is JsonPath or the request type
is JSON. Request
+ // Content-Type (e.g. FORM URLENCODED) does not dictate the response
payload format.
+ if (shouldParseResponseAsJson(appType, jsonOrXPathExpr)) {
Object raw =
JsonPath.using(JSON_PATH_CONFIGURATION).parse(NVL(body,
"{}")).read(jsonOrXPathExpr);
appendStructuredJsonPieces(out, raw);
@@ -960,7 +977,9 @@ public class Rest extends BaseTransform<RestMeta, RestData>
{
}
throw new HopException(
- "REST resultSplitPath requires application type XML or JSON, got '"
+ appType + "'.");
+ "REST resultSplitPath requires a JsonPath (starting with $) or XML
application type, got '"
+ + appType
+ + "'.");
} catch (HopException he) {
throw he;
@@ -1006,14 +1025,13 @@ public class Rest extends BaseTransform<RestMeta,
RestData> {
}
String appType = NVL(resolve(meta.getApplicationType()), "");
try {
- if (RestMeta.APPLICATION_TYPE_JSON.equals(appType)) {
- if (Utils.isEmpty(conn.getCursorJsonPath())) {
- return java.util.Optional.empty();
- }
+ String cursorJsonPath = conn.getCursorJsonPath();
+ if (shouldParseResponseAsJson(appType, resolve(cursorJsonPath))
+ && !Utils.isEmpty(cursorJsonPath)) {
Object raw =
JsonPath.using(JSON_PATH_CONFIGURATION)
.parse(NVL(body, "{}"))
- .read(resolve(conn.getCursorJsonPath()));
+ .read(resolve(cursorJsonPath));
if (raw == null) {
return java.util.Optional.empty();
}
@@ -1052,14 +1070,13 @@ public class Rest extends BaseTransform<RestMeta,
RestData> {
}
String appType = NVL(resolve(meta.getApplicationType()), "");
try {
- if (RestMeta.APPLICATION_TYPE_JSON.equals(appType)) {
- if (Utils.isEmpty(conn.getNextPageUrlJsonPath())) {
- return java.util.Optional.empty();
- }
+ String nextPageUrlJsonPath = conn.getNextPageUrlJsonPath();
+ if (shouldParseResponseAsJson(appType, resolve(nextPageUrlJsonPath))
+ && !Utils.isEmpty(nextPageUrlJsonPath)) {
Object raw =
JsonPath.using(JSON_PATH_CONFIGURATION)
.parse(NVL(body, "{}"))
-
.read(resolveSplitPathOrPagingExpression(conn.getNextPageUrlJsonPath()));
+ .read(resolveSplitPathOrPagingExpression(nextPageUrlJsonPath));
return normalizeNextPageUrl(raw);
}
diff --git
a/plugins/transforms/rest/src/test/java/org/apache/hop/pipeline/transforms/rest/ResponseJsonParseTest.java
b/plugins/transforms/rest/src/test/java/org/apache/hop/pipeline/transforms/rest/ResponseJsonParseTest.java
new file mode 100644
index 0000000000..bf7e8e8c23
--- /dev/null
+++
b/plugins/transforms/rest/src/test/java/org/apache/hop/pipeline/transforms/rest/ResponseJsonParseTest.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.rest;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import org.apache.hop.core.logging.ILoggingObject;
+import org.apache.hop.pipeline.transforms.mock.TransformMockHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Response body parsing for pagination/split must not be locked to the
request Content-Type.
+ * Slack-style FORM posts still return JSON responses addressed by JsonPath.
+ */
+class ResponseJsonParseTest {
+
+ private TransformMockHelper<RestMeta, RestData> mockHelper;
+ private Rest rest;
+
+ @BeforeEach
+ void setUp() {
+ mockHelper = new TransformMockHelper<>("REST form+jsonpath",
RestMeta.class, RestData.class);
+ when(mockHelper.logChannelFactory.create(any(), any(ILoggingObject.class)))
+ .thenReturn(mockHelper.iLogChannel);
+
+ RestMeta meta = new RestMeta();
+ meta.setApplicationType(RestMeta.APPLICATION_TYPE_FORM_URLENCODED);
+ RestData data = new RestData();
+ rest =
+ new Rest(
+ mockHelper.transformMeta, meta, data, 0, mockHelper.pipelineMeta,
mockHelper.pipeline);
+ }
+
+ @AfterEach
+ void tearDown() {
+ mockHelper.cleanUp();
+ }
+
+ @Test
+ void shouldParseResponseAsJsonWhenAppTypeIsJson() {
+ assertTrue(Rest.shouldParseResponseAsJson(RestMeta.APPLICATION_TYPE_JSON,
"not-a-path"));
+ }
+
+ @Test
+ void shouldParseResponseAsJsonWhenPathIsJsonPath() {
+ assertTrue(
+
Rest.shouldParseResponseAsJson(RestMeta.APPLICATION_TYPE_FORM_URLENCODED,
"$.channels[*]"));
+ }
+
+ @Test
+ void shouldNotParseAsJsonWithoutJsonAppTypeOrJsonPath() {
+
assertFalse(Rest.shouldParseResponseAsJson(RestMeta.APPLICATION_TYPE_FORM_URLENCODED,
"/item"));
+
assertFalse(Rest.shouldParseResponseAsJson(RestMeta.APPLICATION_TYPE_TEXT_PLAIN,
null));
+ }
+
+ @Test
+ void formUrlEncodedWithJsonPathAllowsEmptyPagedSplitCheck() {
+ assertTrue(
+ rest.isLegitimateEmptyPagedSplit("{\"channels\":[]}", "$.channels[*]"),
+ "FORM request + JsonPath split on empty JSON array must be treated as
legitimate empty");
+ }
+}
