You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2026/09/09 04:43:10 UTC
[L] Change in asterixdb[master]: [NO ISSUE][TX] Fix data loss and hangs in atomic statement commit
From Ali Alsuliman <al...@gmail.com>:
Ali Alsuliman has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21724?usp=email )
Change subject: [NO ISSUE][TX] Fix data loss and hangs in atomic statement commit
......................................................................
[NO ISSUE][TX] Fix data loss and hangs in atomic statement commit
- user model changes: no
- storage format changes: no
- interface changes: yes
Details:
GlobalTransactionContext.nodeResourceMap was a plain HashMap mutated
without synchronization by handleJobPreparedMessage, which the cluster
controller dispatches on one thread per message. A concurrent put lost a
node's entry, so sendJobCommitMessages skipped that node: its partitions
never committed and their records were lost, while the statement still
reported success after stalling for the whole commit timeout.
Fixed here, with the rest of how the protocol handles the same
acknowledgements:
- make nodeResourceMap concurrent and accumulate each partition's
prepared resources atomically
- test both waits' predicates while holding the monitor, and fail the
statement when the commit wait expires instead of reporting success
- open each round with beginPhase(phase, expectedAcks) taken from a
snapshot of the nodes actually messaged, and drop acknowledgements
belonging to a phase the transaction has already left
- roll back whatever the nodes recorded rather than only what a PREPARED
status implies, so a statement cancelled between job completion and
the commit does not strand flushed components on those nodes
- guarantee removal from txnContextRepository on the paths that own it
- assign the job id before the part of runTrackJob that can throw, so a
cancelled request still aborts and rolls back
Ext-ref: MB-73830
Co-Authored-By: Claude Opus 5 <no...@anthropic.com>
Change-Id: I2eff3bf5c1cc4349b2926a5c1b37b9a3dabc5a24
---
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/GlobalTxManager.java
M asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
A asterixdb/asterix-app/src/test/java/org/apache/asterix/app/cc/GlobalTxManagerTest.java
M asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IGlobalTransactionContext.java
M asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/GlobalTransactionContext.java
A asterixdb/asterix-transactions/src/test/java/org/apache/asterix/transaction/management/service/transaction/GlobalTransactionContextTest.java
6 files changed, 667 insertions(+), 52 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/24/21724/1
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/GlobalTxManager.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/GlobalTxManager.java
index 2ae8a0a..4a5e912 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/GlobalTxManager.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/cc/GlobalTxManager.java
@@ -20,10 +20,12 @@
import static org.apache.hyracks.util.ExitUtil.EC_FAILED_TO_ROLLBACK_ATOMIC_STATEMENT;
+import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
import org.apache.asterix.app.message.AtomicJobCommitMessage;
import org.apache.asterix.app.message.AtomicJobRollbackMessage;
@@ -32,6 +34,7 @@
import org.apache.asterix.common.exceptions.ACIDException;
import org.apache.asterix.common.messaging.api.ICCMessageBroker;
import org.apache.asterix.common.transactions.IGlobalTransactionContext;
+import org.apache.asterix.common.transactions.IGlobalTransactionContext.TxnPhase;
import org.apache.asterix.common.utils.AsterixJobProperty;
import org.apache.asterix.common.utils.StorageConstants;
import org.apache.asterix.transaction.management.service.transaction.GlobalTransactionContext;
@@ -44,10 +47,10 @@
import org.apache.hyracks.api.job.JobStatus;
import org.apache.hyracks.api.job.resource.IJobCapacityController;
import org.apache.hyracks.control.cc.ClusterControllerService;
-import org.apache.hyracks.control.common.controllers.CCConfig;
import org.apache.hyracks.control.nc.io.IOManager;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
import org.apache.hyracks.util.ExitUtil;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -73,32 +76,50 @@
}
@Override
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "Guarded both waits against lost wakeups and failed the statement on a commit timeout")
public void commitTransaction(JobId jobId) throws ACIDException {
IGlobalTransactionContext context = getTransactionContext(jobId);
- if (context.getAcksReceived() != context.getNumPartitions()) {
+ try {
+ // the predicate must be tested while holding the monitor; testing it outside loses a notifyAll
+ // issued between the test and the wait, which hangs the statement for good
+ // TODO: bound this wait. Unlike the commit phase below it waits forever, so a node that dies
+ // after reporting some of its partitions but not the rest hangs the statement indefinitely.
+ // Bounding it needs a decision on which timeout governs the prepare phase; reusing
+ // GLOBAL_TXN_COMMIT_TIMEOUT would silently repurpose an option named for the other phase.
synchronized (context) {
- try {
+ while (context.getAcksReceived() != context.getNumPartitions()) {
context.wait();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new ACIDException(e);
}
}
- }
- context.setTxnStatus(TransactionStatus.PREPARED);
- context.persist(ioManager);
- context.resetAcksReceived();
- sendJobCommitMessages(context);
+ context.setTxnStatus(TransactionStatus.PREPARED);
+ context.persist(ioManager);
- synchronized (context) {
- try {
- CCConfig config = ((ClusterControllerService) serviceContext.getControllerService()).getCCConfig();
- context.wait(config.getGlobalTxCommitTimeout());
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new ACIDException(e);
+ // every participating node reports at least one partition, so anything less means a prepared
+ // message was dropped and committing now would leave the missing node's partitions behind
+ int preparedNodes = context.getNodeResourceMap().size();
+ if (preparedNodes < context.getNumNodes()) {
+ throw new ACIDException("Prepared resources of " + jobId + " cover " + preparedNodes + " node(s) but "
+ + context.getNumNodes() + " node(s) participated in the job");
}
+
+ sendJobCommitMessages(context);
+
+ long timeout = ((ClusterControllerService) serviceContext.getControllerService()).getCCConfig()
+ .getGlobalTxCommitTimeout();
+ awaitStatus(context, TransactionStatus.COMMITTED, timeout);
+ if (context.getTxnStatus() != TransactionStatus.COMMITTED) {
+ throw new ACIDException("Timed out after " + timeout + "ms waiting for " + context.getExpectedAcks()
+ + " node(s) to commit " + jobId + "; " + context.getAcksReceived()
+ + " acknowledged. The transaction log is retained for rollback");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ACIDException(e);
}
+ // deliberately not in a finally: on failure the context is handed to the caller, which aborts the
+ // transaction, and abortTransaction has to be able to look it up to roll the prepared nodes back.
+ // Removing it here would turn every failure into "Transaction for jobId ... does not exist" and skip
+ // the rollback, stranding the flushed components on the nodes.
txnContextRepository.remove(jobId);
}
@@ -112,6 +133,7 @@
}
@Override
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "Accumulate the reported resources atomically instead of racing on a plain map")
public void handleJobPreparedMessage(JobId jobId, String nodeId, Map<String, ILSMComponentId> componentIdMap) {
IGlobalTransactionContext context = txnContextRepository.get(jobId);
if (context == null) {
@@ -119,11 +141,15 @@
+ ", which does not exist. The transaction for the job is already aborted");
return;
}
- if (context.getNodeResourceMap().containsKey(nodeId)) {
- context.getNodeResourceMap().get(nodeId).putAll(componentIdMap);
- } else {
- context.getNodeResourceMap().put(nodeId, componentIdMap);
+ if (context.getPhase() != TxnPhase.PREPARE) {
+ // the prepare phase ends only once every participating partition has reported, so anything
+ // arriving later is a duplicate. Accepting it would add to the map the next phase has already
+ // snapshotted and inflate the acknowledgement counter, acking that phase a node early.
+ LOGGER.warn("ignoring late JobPreparedMessage from {} for {}, which is in phase {}", nodeId, jobId,
+ context.getPhase());
+ return;
}
+ context.addPreparedNodeResources(nodeId, componentIdMap);
if (context.incrementAndGetAcksReceived() == context.getNumPartitions()) {
synchronized (context) {
context.notifyAll();
@@ -131,8 +157,12 @@
}
}
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "Send to a snapshot of the recipients and record how many acks that phase awaits")
private void sendJobCommitMessages(IGlobalTransactionContext context) {
- for (String nodeId : context.getNodeResourceMap().keySet()) {
+ // snapshot: the nodes messaged and the ack target taken from them must be one set, not two reads
+ List<String> nodeIds = new ArrayList<>(context.getNodeResourceMap().keySet());
+ context.beginPhase(TxnPhase.COMMIT, nodeIds.size());
+ for (String nodeId : nodeIds) {
AtomicJobCommitMessage message = new AtomicJobCommitMessage(context.getJobId(), context.getDatasetIds());
try {
((ICCMessageBroker) serviceContext.getMessageBroker()).sendRealTimeApplicationMessageToNC(message,
@@ -144,9 +174,13 @@
}
@Override
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "Await the snapshot of nodes actually messaged rather than a live map or the job's node count")
public void handleJobCompletionMessage(JobId jobId, String nodeId) {
IGlobalTransactionContext context = getTransactionContext(jobId);
- if (context.incrementAndGetAcksReceived() == context.getNumNodes()) {
+ if (rejectStrayAck(context, TxnPhase.COMMIT, jobId, nodeId, "JobCompletionMessage")) {
+ return;
+ }
+ if (context.incrementAndGetAcksReceived() == context.getExpectedAcks()) {
context.delete(ioManager);
context.setTxnStatus(TransactionStatus.COMMITTED);
synchronized (context) {
@@ -157,9 +191,13 @@
}
@Override
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "Await the snapshot of nodes actually messaged rather than a live map or the job's node count")
public void handleJobRollbackCompletionMessage(JobId jobId, String nodeId) {
IGlobalTransactionContext context = getTransactionContext(jobId);
- if (context.incrementAndGetAcksReceived() == context.getNumNodes()) {
+ if (rejectStrayAck(context, TxnPhase.ROLLBACK, jobId, nodeId, "JobRollbackCompletionMessage")) {
+ return;
+ }
+ if (context.incrementAndGetAcksReceived() == context.getExpectedAcks()) {
context.setTxnStatus(TransactionStatus.ROLLBACK);
context.delete(ioManager);
synchronized (context) {
@@ -186,13 +224,18 @@
public void rollback() throws Exception {
Set<FileReference> txnLogFileRefs = ioManager.list(ioManager.resolve(StorageConstants.GLOBAL_TXN_DIR_NAME));
for (FileReference txnLogFileRef : txnLogFileRefs) {
+ IGlobalTransactionContext context = null;
try {
- IGlobalTransactionContext context = new GlobalTransactionContext(txnLogFileRef, ioManager);
+ context = new GlobalTransactionContext(txnLogFileRef, ioManager);
txnContextRepository.put(context.getJobId(), context);
sendJobRollbackMessages(context);
} catch (Exception e) {
LOGGER.error("Error rolling back transaction for {}", txnLogFileRef, e);
cleanup(txnLogFileRef);
+ } finally {
+ if (context != null) {
+ txnContextRepository.remove(context.getJobId());
+ }
}
}
}
@@ -208,35 +251,57 @@
}
}
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "Guarded the rollback wait with its predicate and reported the timeout instead of assuming success")
private void sendJobRollbackMessages(IGlobalTransactionContext context) throws Exception {
JobId jobId = context.getJobId();
- for (String nodeId : context.getNodeResourceMap().keySet()) {
+ // snapshot: the nodes messaged and the ack target taken from them must be one set, not two reads
+ List<String> nodeIds = new ArrayList<>(context.getNodeResourceMap().keySet());
+ context.beginPhase(TxnPhase.ROLLBACK, nodeIds.size());
+ for (String nodeId : nodeIds) {
AtomicJobRollbackMessage rollbackMessage = new AtomicJobRollbackMessage(jobId, context.getDatasetIds(),
context.getNodeResourceMap().get(nodeId));
((ICCMessageBroker) serviceContext.getMessageBroker()).sendRealTimeApplicationMessageToNC(rollbackMessage,
nodeId);
}
- synchronized (context) {
- try {
- CCConfig config = ((ClusterControllerService) serviceContext.getControllerService()).getCCConfig();
- context.wait(config.getGlobalTxRollbackTimeout());
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- LOGGER.error("Error while rolling back atomic statement for {}, halting JVM", jobId);
- ExitUtil.halt(EC_FAILED_TO_ROLLBACK_ATOMIC_STATEMENT);
- }
+ long timeout = ((ClusterControllerService) serviceContext.getControllerService()).getCCConfig()
+ .getGlobalTxRollbackTimeout();
+ try {
+ awaitStatus(context, TransactionStatus.ROLLBACK, timeout);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.error("Error while rolling back atomic statement for {}, halting JVM", jobId);
+ ExitUtil.halt(EC_FAILED_TO_ROLLBACK_ATOMIC_STATEMENT);
}
- txnContextRepository.remove(jobId);
+ if (context.getTxnStatus() != TransactionStatus.ROLLBACK) {
+ // deliberately reported rather than rethrown: the caller in rollback() treats a failure here as a
+ // corrupted log and deletes it, which would discard the only record of what is left to undo
+ LOGGER.error(
+ "Timed out after {}ms waiting for {} node(s) to roll back {}; {} acknowledged. The "
+ + "transaction log is retained for recovery",
+ timeout, context.getExpectedAcks(), jobId, context.getAcksReceived());
+ }
}
@Override
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "Drop the context even when the rollback messages fail to go out")
public void abortTransaction(JobId jobId) throws Exception {
- IGlobalTransactionContext context = getTransactionContext(jobId);
- context.resetAcksReceived();
- if (context.getTxnStatus() == TransactionStatus.PREPARED) {
- sendJobRollbackMessages(context);
+ try {
+ IGlobalTransactionContext context = getTransactionContext(jobId);
+ // the status cannot be the test here. The nodes flush their components and report them before
+ // commitTransaction moves the transaction to PREPARED, so a statement that fails in between - a
+ // cancellation between waitForCompletion and the commit, say - leaves a fully prepared
+ // transaction still reading ACTIVE, and a partially prepared one never reaches PREPARED at all.
+ // Either way the nodes are holding components that only a rollback message will discard, and the
+ // recorded resources are what say so. COMMITTED and ROLLBACK cannot be seen here: both paths
+ // remove the context, so the lookup above would have thrown first.
+ if (!context.getNodeResourceMap().isEmpty()) {
+ sendJobRollbackMessages(context);
+ }
+ } finally {
+ // nothing else ever removes it: the repository is filled from the job lifecycle but drained only
+ // from the statement path, so a throw on the way out would strand the entry for the CC's lifetime
+ txnContextRepository.remove(jobId);
}
- txnContextRepository.remove(jobId);
}
@Override
@@ -258,4 +323,37 @@
throws HyracksException {
}
+
+ /**
+ * An acknowledgement that belongs to a phase the transaction has already left must be dropped, not
+ * counted: the counter is shared across phases, so a straggler would push the current phase over its
+ * target early and declare it complete while a node has yet to answer.
+ */
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.GENERATED, notes = "Drop acknowledgements belonging to an abandoned phase")
+ private static boolean rejectStrayAck(IGlobalTransactionContext context, TxnPhase expected, JobId jobId,
+ String nodeId, String messageKind) {
+ if (context.getPhase() != expected) {
+ LOGGER.warn("ignoring {} from {} for {}: expected phase {} but the transaction is in phase {}", messageKind,
+ nodeId, jobId, expected, context.getPhase());
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Waits until the acknowledgements of every messaged node have moved the transaction to {@code target}, or
+ * the timeout expires. The caller must re-read the status to tell those two outcomes apart.
+ */
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.GENERATED, notes = "Deadline-bounded, predicate-guarded wait shared by the commit and rollback phases")
+ private static void awaitStatus(IGlobalTransactionContext context, TransactionStatus target, long timeoutMillis)
+ throws InterruptedException {
+ long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMillis);
+ synchronized (context) {
+ long remaining = timeoutMillis;
+ while (context.getTxnStatus() != target && remaining > 0) {
+ context.wait(remaining);
+ remaining = TimeUnit.NANOSECONDS.toMillis(deadline - System.nanoTime());
+ }
+ }
+ }
}
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
index 1a7be70..d05a527 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/app/translator/QueryTranslator.java
@@ -317,6 +317,7 @@
import org.apache.hyracks.storage.am.lsm.invertedindex.fulltext.TokenizerCategory;
import org.apache.hyracks.util.LogRedactionUtil;
import org.apache.hyracks.util.OptionalBoolean;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -4606,6 +4607,7 @@
final ClientRequest clientRequest = (ClientRequest) requestTracker.get(reqId);
jobId = runTrackJob(hcc, spec, jobFlags, reqId, requestParameters.getClientContextId(), clientRequest,
JobKind.DML);
+ trackSubmittedJob(hcc, jobId, clientRequest, reqId);
clientRequest.markCancellable();
String nameBefore = Thread.currentThread().getName();
try {
@@ -4811,6 +4813,7 @@
}
jobId = runTrackJob(hcc, jobSpec, jobFlags, reqId, reqParams.getClientContextId(), clientRequest,
JobKind.DML);
+ trackSubmittedJob(hcc, jobId, clientRequest, reqId);
clientRequest.markCancellable();
String nameBefore = Thread.currentThread().getName();
try {
@@ -4883,6 +4886,7 @@
final ClientRequest clientRequest = (ClientRequest) requestTracker.get(reqId);
jobId = runTrackJob(hcc, jobSpec, jobFlags, reqId, requestParameters.getClientContextId(),
clientRequest, JobKind.DML);
+ trackSubmittedJob(hcc, jobId, clientRequest, reqId);
clientRequest.markCancellable();
String nameBefore = Thread.currentThread().getName();
try {
@@ -4913,17 +4917,28 @@
private static JobId runTrackJob(IHyracksClientConnection hcc, JobSpecification jobSpec, EnumSet<JobFlag> jobFlags,
String reqId, String clientCtxId, ClientRequest clientRequest, JobKind jobKind) throws Exception {
// Guard before submitting the job: if the request was cancelled and removed from the tracker,
- // clientRequest will be null here; treat that as a cancellation rather than NPE on addJob.
+ // clientRequest will be null here; treat that as a cancellation rather than NPE in trackSubmittedJob.
ensureNotCancelled(clientRequest, reqId);
jobSpec.setRequestId(reqId);
jobSpec.setProperty(JOB_KIND, jobKind);
JobId jobId = JobUtils.runJobIfActive(hcc, jobSpec, jobFlags, false);
LOGGER.info("Created job {} for uuid:{}, clientContextID:{}", jobId, reqId, clientCtxId);
+ return jobId;
+ }
+
+ /**
+ * Attaches a submitted job to its request, cancelling the job if the request is already gone. Split out of
+ * {@link #runTrackJob} so that the caller holds the job id before anything here can throw: creating the
+ * job registers an atomic statement's global transaction on the cluster controller, and the caller's job
+ * id is the only handle by which that transaction can then be rolled back and deregistered.
+ */
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "Split off so a cancellation after submission cannot strand the global transaction")
+ private static void trackSubmittedJob(IHyracksClientConnection hcc, JobId jobId, ClientRequest clientRequest,
+ String reqId) throws Exception {
if (!clientRequest.addJob(jobId)) {
hcc.cancelJob(jobId);
throw new RuntimeDataException(ErrorCode.REQUEST_CANCELLED, reqId);
}
- return jobId;
}
@Override
@@ -6370,6 +6385,7 @@
}
jobId = runTrackJob(hcc, jobSpec, jobFlags, reqId, requestParameters.getClientContextId(), clientRequest,
jobKind);
+ trackSubmittedJob(hcc, jobId, clientRequest, reqId);
if (jId != null) {
jId.setValue(jobId);
}
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/cc/GlobalTxManagerTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/cc/GlobalTxManagerTest.java
new file mode 100644
index 0000000..d98d081
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/app/cc/GlobalTxManagerTest.java
@@ -0,0 +1,333 @@
+/*
+ * 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.asterix.app.cc;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.asterix.app.message.AtomicJobCommitMessage;
+import org.apache.asterix.app.message.AtomicJobRollbackMessage;
+import org.apache.asterix.common.cluster.IGlobalTxManager.TransactionStatus;
+import org.apache.asterix.common.exceptions.ACIDException;
+import org.apache.asterix.common.messaging.api.ICCMessageBroker;
+import org.apache.asterix.common.messaging.api.INcAddressedMessage;
+import org.apache.asterix.common.transactions.IGlobalTransactionContext;
+import org.apache.hyracks.api.application.ICCServiceContext;
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.control.cc.ClusterControllerService;
+import org.apache.hyracks.control.common.controllers.CCConfig;
+import org.apache.hyracks.control.nc.io.IOManager;
+import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentId;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Covers the two ways the rollback phase can end: every node acknowledges, or the wait expires. Both used to
+ * be indistinguishable, because the wait was a bare timed {@code wait()} with no predicate and no check of
+ * what it woke up to.
+ */
+@AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED, notes = "Covers rollback completion and rollback timeout of the atomic statement protocol")
+public class GlobalTxManagerTest {
+
+ private static final long ROLLBACK_TIMEOUT_MILLIS = 500;
+ private static final long COMMIT_TIMEOUT_MILLIS = 500;
+ private static final String NODE_ID = "node_0";
+ private static final int DATASET_ID = 101;
+
+ private ICCServiceContext serviceContext;
+ private ICCMessageBroker messageBroker;
+ private IOManager ioManager;
+ private GlobalTxManager globalTxManager;
+
+ @Before
+ public void setUp() {
+ CCConfig ccConfig = mock(CCConfig.class);
+ when(ccConfig.getGlobalTxRollbackTimeout()).thenReturn(ROLLBACK_TIMEOUT_MILLIS);
+ when(ccConfig.getGlobalTxCommitTimeout()).thenReturn(COMMIT_TIMEOUT_MILLIS);
+ ClusterControllerService controllerService = mock(ClusterControllerService.class);
+ when(controllerService.getCCConfig()).thenReturn(ccConfig);
+ messageBroker = mock(ICCMessageBroker.class);
+ serviceContext = mock(ICCServiceContext.class);
+ when(serviceContext.getControllerService()).thenReturn(controllerService);
+ when(serviceContext.getMessageBroker()).thenReturn(messageBroker);
+ ioManager = mock(IOManager.class);
+ globalTxManager = new GlobalTxManager(serviceContext, ioManager);
+ }
+
+ /**
+ * The acknowledgement can land while the rollback messages are still being sent, i.e. before the sender
+ * reaches its wait. Testing the predicate only after waiting misses that notify and blocks for the whole
+ * timeout, so this asserts the abort returns well inside it.
+ */
+ @Test
+ public void rollbackAcknowledgedBeforeWaitDoesNotBlock() throws Exception {
+ JobId jobId = new JobId(1);
+ prepareTransaction(jobId);
+ when(messageBroker.sendRealTimeApplicationMessageToNC(any(INcAddressedMessage.class), anyString()))
+ .thenAnswer(invocation -> {
+ globalTxManager.handleJobRollbackCompletionMessage(jobId, NODE_ID);
+ return true;
+ });
+
+ long elapsed = timeAbort(jobId);
+
+ assertTrue("the abort waited for the timeout despite being acknowledged, took " + elapsed + "ms",
+ elapsed < ROLLBACK_TIMEOUT_MILLIS);
+ verify(ioManager).delete(any());
+ }
+
+ /**
+ * When no node acknowledges, the rollback must not be mistaken for a completed one: the transaction log is
+ * the only record of what is still to be undone, so it has to survive for startup recovery.
+ * <p>
+ * This pins the invariants rather than reproducing a past defect: the timed-out rollback was always
+ * observably like this, the difference is that it is now reported instead of passing unnoticed, and the
+ * report is a log line no assertion here can see. What it does catch is a future change that deletes the
+ * log on timeout, flips the status as though the rollback had finished, or returns before it expires.
+ */
+ @Test
+ public void rollbackTimeoutRetainsTransactionLog() throws Exception {
+ JobId jobId = new JobId(2);
+ IGlobalTransactionContext context = prepareTransaction(jobId);
+ when(messageBroker.sendRealTimeApplicationMessageToNC(any(INcAddressedMessage.class), anyString()))
+ .thenReturn(true);
+
+ long elapsed = timeAbort(jobId);
+
+ assertTrue("the abort returned before the rollback timeout expired, took " + elapsed + "ms",
+ elapsed >= ROLLBACK_TIMEOUT_MILLIS);
+ assertNotEquals("a timed-out rollback must not be reported as rolled back", TransactionStatus.ROLLBACK,
+ context.getTxnStatus());
+ verify(ioManager, never()).delete(any());
+ }
+
+ /**
+ * A duplicate prepared message can still arrive once the prepare phase has closed. The acknowledgement
+ * counter is shared by both phases, so accepting one inflates the commit count and makes the next genuine
+ * acknowledgement look like the last: the transaction would be reported committed, and its log deleted,
+ * while a node that never acknowledged is still holding uncommitted components.
+ */
+ @Test
+ public void latePreparedMessageDoesNotCommitEarly() throws Exception {
+ JobId jobId = new JobId(3);
+ String secondNode = "node_1";
+ IGlobalTransactionContext context =
+ globalTxManager.beginTransaction(jobId, 2, 2, Collections.singletonList(DATASET_ID));
+ globalTxManager.handleJobPreparedMessage(jobId, NODE_ID,
+ Collections.singletonMap("resource_0", new LSMComponentId(1, 1)));
+ globalTxManager.handleJobPreparedMessage(jobId, secondNode,
+ Collections.singletonMap("resource_1", new LSMComponentId(1, 1)));
+
+ // while the commit messages go out, NODE_ID repeats its prepared message and then acknowledges the
+ // commit. node_1 never acknowledges, so one of the two commits is still outstanding throughout.
+ AtomicInteger sends = new AtomicInteger();
+ when(messageBroker.sendRealTimeApplicationMessageToNC(any(INcAddressedMessage.class), anyString()))
+ .thenAnswer(invocation -> {
+ if (sends.getAndIncrement() == 0) {
+ globalTxManager.handleJobPreparedMessage(jobId, NODE_ID,
+ Collections.singletonMap("resource_0", new LSMComponentId(1, 1)));
+ globalTxManager.handleJobCompletionMessage(jobId, NODE_ID);
+ }
+ return true;
+ });
+
+ try {
+ globalTxManager.commitTransaction(jobId);
+ fail("the commit was reported as successful while one node had not acknowledged it");
+ } catch (ACIDException e) {
+ assertTrue("expected a commit timeout, got: " + e.getMessage(),
+ e.getMessage() != null && e.getMessage().contains("Timed out"));
+ }
+ assertNotEquals("an incomplete commit must not be reported as committed", TransactionStatus.COMMITTED,
+ context.getTxnStatus());
+ verify(ioManager, never()).delete(any());
+ }
+
+ /**
+ * A commit that times out is followed by a rollback, and the slow node's commit acknowledgement can still
+ * turn up while the rollback is in flight. The two phases share one counter, so counting that straggler
+ * would let a single rollback acknowledgement satisfy a two-node target: the rollback would be declared
+ * complete and its log deleted while the other node had not rolled back at all.
+ */
+ @Test
+ public void lateCommitAckDoesNotCompleteRollbackEarly() throws Exception {
+ JobId jobId = new JobId(4);
+ String secondNode = "node_1";
+ IGlobalTransactionContext context =
+ globalTxManager.beginTransaction(jobId, 2, 2, Collections.singletonList(DATASET_ID));
+ globalTxManager.handleJobPreparedMessage(jobId, NODE_ID,
+ Collections.singletonMap("resource_0", new LSMComponentId(1, 1)));
+ globalTxManager.handleJobPreparedMessage(jobId, secondNode,
+ Collections.singletonMap("resource_1", new LSMComponentId(1, 1)));
+
+ AtomicInteger commitSends = new AtomicInteger();
+ AtomicInteger rollbackSends = new AtomicInteger();
+ when(messageBroker.sendRealTimeApplicationMessageToNC(any(INcAddressedMessage.class), anyString()))
+ .thenAnswer(invocation -> {
+ Object message = invocation.getArgument(0);
+ if (message instanceof AtomicJobCommitMessage && commitSends.getAndIncrement() == 0) {
+ // only NODE_ID acknowledges the commit, so it has to time out
+ globalTxManager.handleJobCompletionMessage(jobId, NODE_ID);
+ } else if (message instanceof AtomicJobRollbackMessage && rollbackSends.getAndIncrement() == 0) {
+ // node_1's commit acknowledgement finally arrives, mid-rollback, and NODE_ID rolls
+ // back. node_1 never acknowledges the rollback.
+ globalTxManager.handleJobCompletionMessage(jobId, secondNode);
+ globalTxManager.handleJobRollbackCompletionMessage(jobId, NODE_ID);
+ }
+ return true;
+ });
+
+ try {
+ globalTxManager.commitTransaction(jobId);
+ fail("the commit was reported as successful while one node had not acknowledged it");
+ } catch (ACIDException e) {
+ assertTrue("expected a commit timeout, got: " + e.getMessage(),
+ e.getMessage() != null && e.getMessage().contains("Timed out"));
+ }
+ globalTxManager.abortTransaction(jobId);
+
+ assertNotEquals("an incomplete rollback must not be reported as rolled back", TransactionStatus.ROLLBACK,
+ context.getTxnStatus());
+ verify(ioManager, never()).delete(any());
+ }
+
+ /**
+ * The repository is filled from the job lifecycle but drained only from the statement path, so an abort
+ * that throws on its way out is the last chance to drop the entry. Leaking it strands the context, and
+ * the resource map it holds, for the lifetime of the cluster controller.
+ */
+ @Test
+ public void failedRollbackSendStillDropsTheContext() throws Exception {
+ JobId jobId = new JobId(5);
+ prepareTransaction(jobId);
+ when(messageBroker.sendRealTimeApplicationMessageToNC(any(INcAddressedMessage.class), anyString()))
+ .thenThrow(new IllegalStateException("broker is down"));
+
+ try {
+ globalTxManager.abortTransaction(jobId);
+ fail("the rollback send failed, so the abort should not have reported success");
+ } catch (RuntimeException expected) {
+ // the send failure propagates to the caller, which is what marks the abort as unfinished
+ }
+ assertEquals("the context outlived a failed abort", 0, countTracked(jobId));
+ }
+
+ /**
+ * The nodes flush and report their components before {@code commitTransaction} moves the transaction to
+ * PREPARED, so a statement cancelled in between - after {@code waitForCompletion} succeeds and before the
+ * commit - aborts a transaction that is fully prepared but still reads ACTIVE. It has to be rolled back
+ * anyway: nothing else discards those components. The nodes only self-abort when their own task fails,
+ * and no transaction log has been persisted yet for recovery to act on.
+ */
+ @Test
+ public void abortRollsBackNodesThatPreparedBeforeTheCommitBegan() throws Exception {
+ JobId jobId = new JobId(6);
+ String secondNode = "node_1";
+ IGlobalTransactionContext context =
+ globalTxManager.beginTransaction(jobId, 2, 2, Collections.singletonList(DATASET_ID));
+ globalTxManager.handleJobPreparedMessage(jobId, NODE_ID,
+ Collections.singletonMap("resource_0", new LSMComponentId(1, 1)));
+ globalTxManager.handleJobPreparedMessage(jobId, secondNode,
+ Collections.singletonMap("resource_1", new LSMComponentId(1, 1)));
+ assertEquals("the commit has not run, so the transaction is still active", TransactionStatus.ACTIVE,
+ context.getTxnStatus());
+
+ List<String> rolledBack = new ArrayList<>();
+ when(messageBroker.sendRealTimeApplicationMessageToNC(any(INcAddressedMessage.class), anyString()))
+ .thenAnswer(invocation -> {
+ assertTrue("expected a rollback message, got " + invocation.getArgument(0),
+ invocation.getArgument(0) instanceof AtomicJobRollbackMessage);
+ rolledBack.add(invocation.getArgument(1));
+ globalTxManager.handleJobRollbackCompletionMessage(jobId, invocation.getArgument(1));
+ return true;
+ });
+
+ globalTxManager.abortTransaction(jobId);
+
+ assertEquals("both prepared nodes must be told to discard their components", 2, rolledBack.size());
+ assertTrue("node_0 was not rolled back", rolledBack.contains(NODE_ID));
+ assertTrue("node_1 was not rolled back", rolledBack.contains(secondNode));
+ assertEquals(TransactionStatus.ROLLBACK, context.getTxnStatus());
+ }
+
+ /**
+ * A partially prepared transaction never reaches PREPARED at all, and the partitions that did report are
+ * still holding components. The rollback goes to exactly those, and to no one else.
+ */
+ @Test
+ public void abortRollsBackOnlyThePartitionsThatPrepared() throws Exception {
+ JobId jobId = new JobId(7);
+ globalTxManager.beginTransaction(jobId, 2, 2, Collections.singletonList(DATASET_ID));
+ globalTxManager.handleJobPreparedMessage(jobId, NODE_ID,
+ Collections.singletonMap("resource_0", new LSMComponentId(1, 1)));
+
+ List<String> rolledBack = new ArrayList<>();
+ when(messageBroker.sendRealTimeApplicationMessageToNC(any(INcAddressedMessage.class), anyString()))
+ .thenAnswer(invocation -> {
+ rolledBack.add(invocation.getArgument(1));
+ globalTxManager.handleJobRollbackCompletionMessage(jobId, invocation.getArgument(1));
+ return true;
+ });
+
+ globalTxManager.abortTransaction(jobId);
+
+ assertEquals("only the partition that reported should be rolled back", Collections.singletonList(NODE_ID),
+ rolledBack);
+ }
+
+ private IGlobalTransactionContext prepareTransaction(JobId jobId) {
+ IGlobalTransactionContext context =
+ globalTxManager.beginTransaction(jobId, 1, 1, Collections.singletonList(DATASET_ID));
+ context.addPreparedNodeResources(NODE_ID, Collections.singletonMap("resource_0", new LSMComponentId(1, 1)));
+ context.setTxnStatus(TransactionStatus.PREPARED);
+ return context;
+ }
+
+ private long timeAbort(JobId jobId) throws Exception {
+ long start = System.nanoTime();
+ globalTxManager.abortTransaction(jobId);
+ long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+ assertEquals("the transaction context outlived the abort", 0, countTracked(jobId));
+ return elapsed;
+ }
+
+ private int countTracked(JobId jobId) {
+ try {
+ globalTxManager.getTransactionContext(jobId);
+ return 1;
+ } catch (Exception e) {
+ return 0;
+ }
+ }
+}
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IGlobalTransactionContext.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IGlobalTransactionContext.java
index 5a175ac..38adf7a 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IGlobalTransactionContext.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/transactions/IGlobalTransactionContext.java
@@ -28,6 +28,17 @@
public interface IGlobalTransactionContext {
+ /**
+ * The round of acknowledgements the transaction is currently collecting. The transaction status cannot
+ * stand in for this: it reads {@code PREPARED} throughout both the commit and the rollback phase, so an
+ * acknowledgement belonging to an abandoned phase is otherwise indistinguishable from a current one.
+ */
+ enum TxnPhase {
+ PREPARE,
+ COMMIT,
+ ROLLBACK
+ }
+
JobId getJobId();
int incrementAndGetAcksReceived();
@@ -38,7 +49,17 @@
int getNumPartitions();
- void resetAcksReceived();
+ /**
+ * Opens a round of acknowledgements: records which phase is collecting them and how many it waits for,
+ * captured from the set of nodes actually messaged, and discards the previous phase's count. Must be
+ * called before the first message of the phase is sent, so that a straggler from the phase before is
+ * rejected rather than counted towards this one.
+ */
+ void beginPhase(TxnPhase phase, int expectedAcks);
+
+ TxnPhase getPhase();
+
+ int getExpectedAcks();
void setTxnStatus(IGlobalTxManager.TransactionStatus status);
@@ -48,6 +69,15 @@
Map<String, Map<String, ILSMComponentId>> getNodeResourceMap();
+ /**
+ * Records the resources a participating partition reported in its prepared message. Called concurrently,
+ * once per participating partition, so implementations must accumulate atomically.
+ *
+ * @param nodeId the node the prepared message came from
+ * @param componentIdMap the flushed component of each resource of the reporting partition; may be empty
+ */
+ void addPreparedNodeResources(String nodeId, Map<String, ILSMComponentId> componentIdMap);
+
void persist(IOManager ioManager);
void delete(IOManager ioManager);
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/GlobalTransactionContext.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/GlobalTransactionContext.java
index 4cac9d8..ffebf74 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/GlobalTransactionContext.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/transaction/GlobalTransactionContext.java
@@ -19,9 +19,9 @@
package org.apache.asterix.transaction.management.service.transaction;
import java.nio.file.Paths;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.asterix.common.cluster.IGlobalTxManager.TransactionStatus;
@@ -32,6 +32,7 @@
import org.apache.hyracks.api.job.JobId;
import org.apache.hyracks.control.nc.io.IOManager;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.hyracks.util.annotations.ThreadSafe;
import com.fasterxml.jackson.core.JsonProcessingException;
@@ -41,11 +42,20 @@
public class GlobalTransactionContext implements IGlobalTransactionContext {
protected final JobId jobId;
- private AtomicInteger acksReceived;
+ private final AtomicInteger acksReceived;
private final int numNodes;
- private TransactionStatus status;
+ private volatile TransactionStatus status;
private final List<Integer> datasetIds;
private final int numPartitions;
+ private volatile int expectedAcks;
+ private volatile TxnPhase phase = TxnPhase.PREPARE;
+
+ /**
+ * Concurrent because the prepared messages of the participating partitions are handled by several CC
+ * executor threads at once. A plain {@link java.util.HashMap} loses entries under that concurrency, which
+ * drops a node from the commit broadcast and silently leaves its partitions uncommitted.
+ */
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.REFACTORED, notes = "Made the node resource map concurrent; prepared messages are accumulated from several threads")
private final Map<String, Map<String, ILSMComponentId>> nodeResourceMap;
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
@@ -55,7 +65,7 @@
this.numNodes = numNodes;
this.numPartitions = numPartitions;
this.acksReceived = new AtomicInteger(0);
- this.nodeResourceMap = new HashMap<>();
+ this.nodeResourceMap = new ConcurrentHashMap<>();
this.status = TransactionStatus.ACTIVE;
}
@@ -65,7 +75,7 @@
AtomicTransactionLog.class);
this.jobId = txnLog.getJobId();
this.datasetIds = txnLog.getDatasetIds();
- this.nodeResourceMap = txnLog.getNodeResourceMap();
+ this.nodeResourceMap = concurrentCopy(txnLog.getNodeResourceMap());
this.numNodes = nodeResourceMap.keySet().size();
this.numPartitions = txnLog.getNumPartitions();
this.acksReceived = new AtomicInteger(0);
@@ -99,9 +109,42 @@
return acksReceived.get();
}
+ /**
+ * The phase is published last: an acknowledgement racing with this call is then either rejected against
+ * the previous phase, or counted against a target that is already in place.
+ */
@Override
- public void resetAcksReceived() {
- acksReceived = new AtomicInteger(0);
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.GENERATED, notes = "Open a phase, its acknowledgement target and its counter in one step")
+ public void beginPhase(TxnPhase phase, int expectedAcks) {
+ acksReceived.set(0);
+ this.expectedAcks = expectedAcks;
+ this.phase = phase;
+ }
+
+ @Override
+ public TxnPhase getPhase() {
+ return phase;
+ }
+
+ @Override
+ public int getExpectedAcks() {
+ return expectedAcks;
+ }
+
+ @Override
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.GENERATED, notes = "Atomic accumulation of a partition's prepared resources")
+ public void addPreparedNodeResources(String nodeId, Map<String, ILSMComponentId> componentIdMap) {
+ nodeResourceMap.computeIfAbsent(nodeId, k -> new ConcurrentHashMap<>()).putAll(componentIdMap);
+ }
+
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.GENERATED, notes = "Keep the recovered map concurrent so rollback shares the live map's guarantees")
+ private static Map<String, Map<String, ILSMComponentId>> concurrentCopy(
+ Map<String, Map<String, ILSMComponentId>> source) {
+ Map<String, Map<String, ILSMComponentId>> copy = new ConcurrentHashMap<>();
+ if (source != null) {
+ source.forEach((nodeId, resources) -> copy.put(nodeId, new ConcurrentHashMap<>(resources)));
+ }
+ return copy;
}
public int getNumNodes() {
diff --git a/asterixdb/asterix-transactions/src/test/java/org/apache/asterix/transaction/management/service/transaction/GlobalTransactionContextTest.java b/asterixdb/asterix-transactions/src/test/java/org/apache/asterix/transaction/management/service/transaction/GlobalTransactionContextTest.java
new file mode 100644
index 0000000..b3bbaba
--- /dev/null
+++ b/asterixdb/asterix-transactions/src/test/java/org/apache/asterix/transaction/management/service/transaction/GlobalTransactionContextTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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.asterix.transaction.management.service.transaction;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentId;
+import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentId;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.junit.Test;
+
+@AiProvenance(agent = AiProvenance.Agent.CLAUDE_OPUS_5, tool = AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind = AiProvenance.ContributionKind.TEST_GENERATED, notes = "Regression test for prepared resources lost when several partitions report concurrently")
+public class GlobalTransactionContextTest {
+
+ private static final int NUM_NODES = 8;
+ private static final int PARTITIONS_PER_NODE = 4;
+ private static final int NUM_PARTITIONS = NUM_NODES * PARTITIONS_PER_NODE;
+ private static final int ITERATIONS = 200;
+
+ /**
+ * Every participating partition reports its prepared resources on its own CC executor thread, and all of
+ * them must survive. A node missing from the map is never sent a commit message, so its partitions stay
+ * uncommitted while the statement still reports success.
+ */
+ @Test
+ public void concurrentPreparedResourcesAreNotLost() throws Exception {
+ ExecutorService executor = Executors.newFixedThreadPool(NUM_PARTITIONS);
+ try {
+ for (int iteration = 0; iteration < ITERATIONS; iteration++) {
+ GlobalTransactionContext context = new GlobalTransactionContext(new JobId(iteration),
+ Collections.singletonList(1), NUM_NODES, NUM_PARTITIONS);
+ CyclicBarrier startLine = new CyclicBarrier(NUM_PARTITIONS);
+ CountDownLatch done = new CountDownLatch(NUM_PARTITIONS);
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ for (int node = 0; node < NUM_NODES; node++) {
+ for (int partition = 0; partition < PARTITIONS_PER_NODE; partition++) {
+ String nodeId = "node_" + node;
+ String resource = nodeId + "/partition_" + partition;
+ executor.submit(() -> {
+ try {
+ startLine.await();
+ context.addPreparedNodeResources(nodeId,
+ Collections.singletonMap(resource, new LSMComponentId(1, 1)));
+ } catch (Throwable th) {
+ failure.compareAndSet(null, th);
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+ }
+ assertTrue("timed out waiting for the partitions to report", done.await(30, TimeUnit.SECONDS));
+ if (failure.get() != null) {
+ throw new AssertionError("reporting a prepared partition failed", failure.get());
+ }
+ Map<String, Map<String, ILSMComponentId>> nodeResourceMap = context.getNodeResourceMap();
+ assertEquals("iteration " + iteration + ": a node was dropped from the commit broadcast", NUM_NODES,
+ nodeResourceMap.size());
+ for (Map.Entry<String, Map<String, ILSMComponentId>> entry : nodeResourceMap.entrySet()) {
+ assertEquals("iteration " + iteration + ": " + entry.getKey() + " lost a partition",
+ PARTITIONS_PER_NODE, entry.getValue().size());
+ }
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+}
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21724?usp=email
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I2eff3bf5c1cc4349b2926a5c1b37b9a3dabc5a24
Gerrit-Change-Number: 21724
Gerrit-PatchSet: 1
Gerrit-Owner: Ali Alsuliman <al...@gmail.com>