This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12600-f7d24f17dc392d80c3e9d2e797d038d7deefecc2 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 623d8691a7e1aa61af59f7834af336b2f9917999 Author: ZIHAN DAI <[email protected]> AuthorDate: Sun Oct 4 22:59:58 2026 +0000 [Fix][Connector-V2] Guard null MongoDB session during committer close (#12600) --- .../commit/MongodbSinkAggregatedCommitter.java | 4 +- .../commit/MongodbSinkAggregatedCommitterTest.java | 43 ++++++++++++++++++++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/commit/MongodbSinkAggregatedCommitter.java b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/commit/MongodbSinkAggregatedCommitter.java index 0ee73a3012..fb0195b61b 100644 --- a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/commit/MongodbSinkAggregatedCommitter.java +++ b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/commit/MongodbSinkAggregatedCommitter.java @@ -153,7 +153,9 @@ public class MongodbSinkAggregatedCommitter @Override public void close() { long deadline = System.currentTimeMillis() + TRANSACTION_TIMEOUT_MS; - while (clientSession.hasActiveTransaction() && System.currentTimeMillis() < deadline) { + while (clientSession != null + && clientSession.hasActiveTransaction() + && System.currentTimeMillis() < deadline) { // wait for active transaction to finish or timeout Thread.sleep(waitingTime); } diff --git a/seatunnel-connectors-v2/connector-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/commit/MongodbSinkAggregatedCommitterTest.java b/seatunnel-connectors-v2/connector-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/commit/MongodbSinkAggregatedCommitterTest.java new file mode 100644 index 0000000000..bcb5076b7f --- /dev/null +++ b/seatunnel-connectors-v2/connector-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/commit/MongodbSinkAggregatedCommitterTest.java @@ -0,0 +1,43 @@ +/* + * 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.seatunnel.connectors.seatunnel.mongodb.sink.commit; + +import org.apache.seatunnel.connectors.seatunnel.mongodb.sink.MongodbWriterOptions; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class MongodbSinkAggregatedCommitterTest { + + @Test + void closeBeforeFirstCommitShouldNotFail() { + MongodbWriterOptions options = + MongodbWriterOptions.builder() + .withConnectString("mongodb://localhost:27017") + .withDatabase("test") + .withCollection("test") + .withTransaction(true) + .build(); + + MongodbSinkAggregatedCommitter committer = new MongodbSinkAggregatedCommitter(options); + + Assertions.assertDoesNotThrow(committer::close); + } +}
