Skip to content

MLE-26420 Improved error handling for incremental writes #1869

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rjrudin merged 1 commit into from
Dec 31, 2025
Merged

MLE-26420 Improved error handling for incremental writes #1869

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion ...nt-api/src/main/java/com/marklogic/client/datamovement/filter/IncrementalWriteFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,9 @@ private String serializeContent(DocumentWriteOperation doc) {
jc = new JsonCanonicalizer(content);
return jc.getEncodedString();
} catch (IOException e) {
// Going to improve this in the next PR, as I think we can throw an exception if Format = JSON.
// If the Format is actually JSON, then the write to MarkLogic should ultimately fail, which is the
// error message the user would want to see via a batch failure listener. So in all cases, if we cannot
// canonicalize something that appears to be JSON, we log a warning and return the original content for hashing.
logger.warn("Unable to canonicalize JSON content for URI {}, using original content for hashing; cause: {}",
doc.getUri(), e.getMessage());
}
Expand Down
94 changes: 41 additions & 53 deletions ...ient-api/src/test/java/com/marklogic/client/datamovement/filter/IncrementalWriteTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,10 @@

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.marklogic.client.DatabaseClient;
import com.marklogic.client.datamovement.DataMovementManager;
import com.marklogic.client.datamovement.WriteBatcher;
import com.marklogic.client.document.DocumentWriteOperation;
import com.marklogic.client.impl.DocumentWriteOperationImpl;
import com.marklogic.client.io.DocumentMetadataHandle;
import com.marklogic.client.io.Format;
import com.marklogic.client.io.JacksonHandle;
import com.marklogic.client.io.StringHandle;
import com.marklogic.client.test.AbstractClientTest;
Expand All @@ -21,9 +19,9 @@
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.concurrent.atomic.AtomicReference;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.*;

class IncrementalWriteTest extends AbstractClientTest {

Expand All @@ -33,23 +31,26 @@ class IncrementalWriteTest extends AbstractClientTest {

AtomicInteger writtenCount = new AtomicInteger();
AtomicInteger skippedCount = new AtomicInteger();
AtomicReference<Throwable> batchFailure = new AtomicReference<>();
ObjectMapper objectMapper = new ObjectMapper();

List<DocumentWriteOperation> docs = new ArrayList<>();
IncrementalWriteFilter filter;

@BeforeEach
void setup() {
Comment on lines +37 to 41
Copy link

Copilot AI Dec 31, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The docs list is now shared across all test methods as instance state. This can lead to test pollution if not properly cleared between tests. Consider initializing it in the @BeforeEach method or making it local to each test method to ensure test isolation.

Suggested change
List<DocumentWriteOperation> docs = new ArrayList<>();
IncrementalWriteFilter filter;
@BeforeEach
void setup() {
List<DocumentWriteOperation> docs;
IncrementalWriteFilter filter;
@BeforeEach
void setup() {
// Ensure docs is reinitialized for each test to avoid shared state between tests.
docs = new ArrayList<>();

Copilot uses AI. Check for mistakes.
// Need a user with eval privileges so that the eval filter can be tested.
Common.client = Common.newEvalClient();
}

@Test
void opticFilter() {
// Default filter implementation, should be suitable for most tests.
filter = IncrementalWriteFilter.newBuilder()
.onDocumentsSkipped(docs -> skippedCount.addAndGet(docs.length))
.build();
}

runTest();
@Test
void opticFilter() {
verifyIncrementalWriteWorks();
}

@Test
Expand All @@ -59,35 +60,11 @@ void evalFilter() {
.onDocumentsSkipped(docs -> skippedCount.addAndGet(docs.length))
.build();

runTest();
}

@Test
void filterRemovesAllDocuments() {
new WriteBatcherTemplate(Common.client).runWriteJob(
writeBatcher -> writeBatcher
.withDocumentWriteSetFilter(context -> context.getDatabaseClient().newDocumentManager().newWriteSet())
.onBatchSuccess(batch -> writtenCount.addAndGet(batch.getItems().length)),

writeBatcher -> {
for (int i = 1; i <= 10; i++) {
writeBatcher.add("/incremental/test/doc-" + i + ".xml", METADATA, new StringHandle("<doc/>"));
}
}
);

assertEquals(0, writtenCount.get(), "No documents should have been written since the filter removed them all. " +
"This test is verifying that no error will occur either when the filter doesn't return any documents.");
assertCollectionSize("incremental-test", 0);
verifyIncrementalWriteWorks();
}

@Test
void jsonKeysOutOfOrder() {
filter = IncrementalWriteFilter.newBuilder()
.onDocumentsSkipped(docs -> skippedCount.addAndGet(docs.length))
.build();

List<DocumentWriteOperation> docs = new ArrayList<>();
for (int i = 1; i <= 10; i++) {
ObjectNode doc = objectMapper.createObjectNode();
doc.put("number", i);
Expand Down Expand Up @@ -146,7 +123,34 @@ void jsonKeysOutOfOrderWithNoCanonicalization() {
assertEquals(0, skippedCount.get());
}

private void runTest() {
@Test
void invalidJsonWithNoFormat() {
docs.add(new DocumentWriteOperationImpl("/not-json.txt", METADATA, new StringHandle("{\"not actually json")));
writeDocs(docs);

assertEquals(1, writtenCount.get(), "When the format is not specified and the content looks like JSON " +
"because it starts with a '{', the JSON canonicalization should fail and log a warning. The " +
"document should still be written with a hash generated based on the text in the document.");

assertNull(batchFailure.get(), "No failure should have been thrown since the format on the content is " +
"not JSON, and thus the content should be hashed as text.");
}

@Test
void invalidJsonWithFormat() {
docs.add(new DocumentWriteOperationImpl("/not.json", METADATA, new StringHandle("not actually json").withFormat(Format.JSON)));
writeDocs(docs);

assertNotNull(batchFailure.get(), "A failure should have been thrown by the server since the content is not " +
"JSON. But the failure to canonicalize should still be logged, as the user will be far more interested " +
"in the error from the server.");

String message = batchFailure.get().getMessage();
assertTrue(message.contains("failed to apply resource at documents"),
"Expecting the server to throw an error. Actual message: " + message);
}

private void verifyIncrementalWriteWorks() {
writeTenDocuments();
assertEquals(10, writtenCount.get());
assertEquals(0, skippedCount.get(), "No docs should have been skipped on the first write.");
Expand Down Expand Up @@ -201,26 +205,10 @@ private void writeDocs(List<DocumentWriteOperation> docs) {
new WriteBatcherTemplate(Common.client).runWriteJob(
writeBatcher -> writeBatcher
.withDocumentWriteSetFilter(filter)
.onBatchSuccess(batch -> writtenCount.addAndGet(batch.getItems().length)),
.onBatchSuccess(batch -> writtenCount.addAndGet(batch.getItems().length))
.onBatchFailure((batch, failure) -> batchFailure.set(failure)),

writeBatcher -> docs.forEach(writeBatcher::add)
);
}

// Experimenting with a template that gets rid of some annoying DMSDK boilerplate.
private record WriteBatcherTemplate(DatabaseClient databaseClient) {

public void runWriteJob(Consumer<WriteBatcher> writeBatcherConfigurer, Consumer<WriteBatcher> writeBatcherUser) {
try (DataMovementManager dmm = databaseClient.newDataMovementManager()) {
WriteBatcher writeBatcher = dmm.newWriteBatcher();
writeBatcherConfigurer.accept(writeBatcher);

dmm.startJob(writeBatcher);
writeBatcherUser.accept(writeBatcher);
writeBatcher.flushAndWait();
writeBatcher.awaitCompletion();
dmm.stopJob(writeBatcher);
}
}
}
}