MLE-26918 Added FilterException to identify filter-related errors

Important for the batch retrier in the Spark connector so it doesn't retry a batch when the filter fails.

Improved the code for closing handles as well in BatchWriter.

Author: rjrudin

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/filter/FilterException.java

@@ -0,0 +1,18 @@
+/*
+ * Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
+ */
+package com.marklogic.client.datamovement.filter;
+
+import com.marklogic.client.datamovement.DataMovementException;
+
+/**
+ * Any exception thrown by execution of a {@code DocumentWriteSetFilter} will be wrapped in this exception and
+ * rethrown by the {@code WriteBatcher}, allowing failure listeners to distinguish filter exceptions from other
+ * exceptions that may occur during batch processing.
+ */
+public class FilterException extends DataMovementException {
+
+	public FilterException(String message, Throwable cause) {
+		super(message, cause);
+	}
+}

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/filter/IncrementalWriteOpticFilter.java

@@ -53,6 +53,10 @@ public DocumentWriteSet apply(Context context) {
 				}
 			);
 
+			if (logger.isDebugEnabled()) {
+				logger.debug("Retrieved {} existing hashes for batch of size {}", existingHashes.size(), uris.length);
+			}
+
 			return filterDocuments(context, uri -> existingHashes.get(uri));
 		} catch (FailedRequestException e) {
 			String message = "Unable to query for existing incremental write hashes; cause: " + e.getMessage();

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/filter/IncrementalWriteViewFilter.java

@@ -48,6 +48,10 @@ public DocumentWriteSet apply(Context context) {
 				}
 			);
 
+			if (logger.isDebugEnabled()) {
+				logger.debug("Retrieved {} existing hashes for batch of size {}", existingHashes.size(), uris.length);
+			}
+
 			return filterDocuments(context, uri -> existingHashes.get(uri));
 		} catch (FailedRequestException e) {
 			String message = "Unable to query for existing incremental write hashes from view " + getConfig().getSchemaName() + "." + getConfig().getViewName() + "; cause: " + e.getMessage();

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/impl/BatchWriter.java

@@ -1,12 +1,14 @@
 /*
- * Copyright (c) 2010-2025 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
+ * Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
  */
 package com.marklogic.client.datamovement.impl;
 
 import com.marklogic.client.datamovement.DocumentWriteSetFilter;
+import com.marklogic.client.datamovement.filter.FilterException;
 import com.marklogic.client.document.DocumentWriteOperation;
 import com.marklogic.client.document.DocumentWriteSet;
 import com.marklogic.client.document.XMLDocumentManager;
+import com.marklogic.client.impl.IoUtil;
 import com.marklogic.client.io.Format;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -25,29 +27,33 @@ public void run() {
 			return;
 		}
 
-		try {
-			logger.trace("Begin write batch {} to forest on host '{}'", batchWriteSet.getBatchNumber(), batchWriteSet.getClient().getHost());
+		logger.trace("Begin write batch {} to forest on host '{}'", batchWriteSet.getBatchNumber(), batchWriteSet.getClient().getHost());
+		DocumentWriteSet documentWriteSet = batchWriteSet.getDocumentWriteSet();
 
-			DocumentWriteSet documentWriteSet = batchWriteSet.getDocumentWriteSet();
-			if (filter != null) {
+		if (filter != null) {
+			try {
 				documentWriteSet = filter.apply(batchWriteSet);
-				if (documentWriteSet == null || documentWriteSet.isEmpty()) {
-					logger.debug("Filter returned empty write set for batch {}, skipping write", batchWriteSet.getBatchNumber());
-					closeAllHandles();
-					return;
-				}
-				batchWriteSet.updateWithFilteredDocumentWriteSet(documentWriteSet);
+			} catch (Exception e) {
+				closeAllHandles();
+				String message = String.format("Unable to apply filter to batch %d; cause: %s", batchWriteSet.getBatchNumber(), e.getMessage());
+				onFailure(new FilterException(message, e));
+				return;
+			}
+			if (documentWriteSet == null || documentWriteSet.isEmpty()) {
+				closeAllHandles();
+				logger.debug("Filter returned empty write set for batch {}, skipping write", batchWriteSet.getBatchNumber());
+				return;
 			}
+			batchWriteSet.updateWithFilteredDocumentWriteSet(documentWriteSet);
+		}
 
+		try {
 			writeDocuments(documentWriteSet);
-
-			// This seems like it should be part of a finally block - but it's able to throw an exception. Which implies
-			// that onFailure() should occur when this fails, which seems odd???
-			closeAllHandles();
-
 			onSuccess();
 		} catch (Throwable t) {
 			onFailure(t);
+		} finally {
+			closeAllHandles();
 		}
 	}
 
@@ -79,21 +85,21 @@ private void onFailure(Throwable t) {
 		}
 	}
 
-	private void closeAllHandles() throws Throwable {
-		Throwable lastThrowable = null;
+	/**
+	 * This used to throw a Throwable... but it's not clear what a user would ever do with that if a content handle
+	 * cannot be closed. Instead, this has been altered to use closeQuietly.
+	 */
+	private void closeAllHandles() {
 		for (DocumentWriteOperation doc : batchWriteSet.getDocumentWriteSet()) {
-			try {
-				if (doc.getContent() instanceof Closeable closeable) {
-					closeable.close();
-				}
-				if (doc.getMetadata() instanceof Closeable closeable) {
-					closeable.close();
-				}
-			} catch (Throwable t) {
-				logger.error("Error closing all handles in BatchWriter", t);
-				lastThrowable = t;
+			if (doc == null) {
+				continue;
+			}
+			if (doc.getContent() instanceof Closeable closeable) {
+				IoUtil.closeQuietly(closeable);
+			}
+			if (doc.getMetadata() instanceof Closeable closeable) {
+				IoUtil.closeQuietly(closeable);
 			}
 		}
-		if (lastThrowable != null) throw lastThrowable;
 	}
 }

marklogic-client-api/src/main/java/com/marklogic/client/impl/IoUtil.java

@@ -1,9 +1,12 @@
 /*
- * Copyright (c) 2010-2025 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
+ * Copyright (c) 2010-2026 Progress Software Corporation and/or its subsidiaries or affiliates. All Rights Reserved.
  */
 package com.marklogic.client.impl;
 
+import org.slf4j.LoggerFactory;
+
 import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
 
@@ -23,4 +26,17 @@ static byte[] streamToBytes(InputStream stream) throws IOException {
 		buffer.flush();
 		return buffer.toByteArray();
 	}
+
+	static void closeQuietly(Closeable closeable) {
+		// Reinvented here as we don't yet have a dependency on a 3rd party library that provides this method, and it's
+		// not worth bringing in a dependency just for this.
+		if (closeable != null) {
+			try {
+				closeable.close();
+			} catch (IOException e) {
+				LoggerFactory.getLogger(IoUtil.class)
+					.warn("Unexpected exception while closing stream: %s".formatted(e.getMessage()), e);
+			}
+		}
+	}
 }

marklogic-client-api/src/test/java/com/marklogic/client/datamovement/filter/IncrementalWriteTest.java

@@ -4,6 +4,7 @@
 package com.marklogic.client.datamovement.filter;
 
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.marklogic.client.FailedRequestException;
 import com.marklogic.client.document.*;
 import com.marklogic.client.impl.DocumentWriteOperationImpl;
 import com.marklogic.client.io.*;
@@ -128,10 +129,22 @@ void noRangeIndexForField() {
 		writeTenDocuments();
 
 		assertNotNull(batchFailure.get());
-		String message = batchFailure.get().getMessage();
-		assertTrue(message.contains("Unable to query for existing incremental write hashes") && message.contains("XDMP-FIELDRIDXNOTFOUND"),
+		assertTrue(batchFailure.get() instanceof FilterException,
+			"If the filter fails, the exception should be wrapped in a FilterException so that WriteBatcher " +
+				"failure listeners can distinguish the difference between a filter failure and a write failure. " +
+				"If the filter fails, there is likely no reason to retry the request. Actual exception class: " + batchFailure.get().getClass());
+
+		FilterException filterException = (FilterException) batchFailure.get();
+		String message = filterException.getMessage();
+		assertTrue(message.startsWith("Unable to apply filter to batch 1; cause: "),
+			"The filter exception message should provide context that the error happened when a filter was applied. " +
+				"Actual message: " + message);
+
+		assertTrue(filterException.getCause() instanceof FailedRequestException);
+		String causeMessage = filterException.getCause().getMessage();
+		assertTrue(causeMessage.contains("Unable to query for existing incremental write hashes") && causeMessage.contains("XDMP-FIELDRIDXNOTFOUND"),
 			"When the user tries to use the incremental write feature without the required range index, we should " +
-				"fail with a helpful error message. Actual message: " + message);
+				"fail with a helpful error message. Actual message: " + causeMessage);
 	}
 
 	@Test