Skip to content

Commit 9e96684

Browse files
jyeminjyemin
committed
Add maxCommitTime support to transactions
JAVA-3296
1 parent 063b2a4 commit 9e96684

File tree

21 files changed

+834
-48
lines changed

21 files changed

+834
-48
lines changed

‎driver-async/src/main/com/mongodb/async/client/ClientSessionImpl.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import static com.mongodb.MongoException.UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL;
3434
import static com.mongodb.assertions.Assertions.isTrue;
3535
import static com.mongodb.assertions.Assertions.notNull;
36+
import static java.util.concurrent.TimeUnit.MILLISECONDS;
3637

3738
class ClientSessionImpl extends BaseClientSessionImpl implements ClientSession {
3839

@@ -124,7 +125,8 @@ public void commitTransaction(final SingleResultCallback<Void> callback) {
124125
boolean alreadyCommitted = commitInProgress || transactionState == TransactionState.COMMITTED;
125126
commitInProgress = true;
126127
executor.execute(new CommitTransactionOperation(transactionOptions.getWriteConcern(), alreadyCommitted)
127-
.recoveryToken(getRecoveryToken()),
128+
.recoveryToken(getRecoveryToken())
129+
.maxCommitTime(transactionOptions.getMaxCommitTime(MILLISECONDS), MILLISECONDS),
128130
readConcern, this,
129131
new SingleResultCallback<Void>() {
130132
@Override

‎driver-async/src/test/functional/com/mongodb/async/client/JsonPoweredCrudTestHelper.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -298,6 +298,9 @@ BsonDocument getAggregateResult(final BsonDocument collectionOptions, final Bson
298298
if (arguments.containsKey("collation")) {
299299
iterable.collation(getCollation(arguments.getDocument("collation")));
300300
}
301+
if (arguments.containsKey("maxTimeMS")) {
302+
iterable.maxTime(arguments.getNumber("maxTimeMS").longValue(), TimeUnit.MILLISECONDS);
303+
}
301304
return toResult(iterable);
302305
}
303306

‎driver-async/src/test/functional/com/mongodb/async/client/TransactionsTest.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@
7272
import static com.mongodb.client.CommandMonitoringTestHelper.assertEventsEquality;
7373
import static com.mongodb.client.CommandMonitoringTestHelper.getExpectedEvents;
7474
import static java.util.Collections.singletonList;
75+
import static java.util.concurrent.TimeUnit.MILLISECONDS;
7576
import static org.junit.Assert.assertEquals;
7677
import static org.junit.Assert.assertFalse;
7778
import static org.junit.Assert.assertNotNull;
@@ -265,6 +266,10 @@ private TransactionOptions createDefaultTransactionOptions(final BsonDocument op
265266
if (defaultTransactionOptionsDocument.containsKey("readPreference")) {
266267
builder.readPreference(helper.getReadPreference(defaultTransactionOptionsDocument));
267268
}
269+
if (defaultTransactionOptionsDocument.containsKey("maxCommitTimeMS")) {
270+
builder.maxCommitTime(defaultTransactionOptionsDocument.getNumber("maxCommitTimeMS").longValue(), MILLISECONDS);
271+
}
272+
268273
}
269274
return builder.build();
270275
}
@@ -468,6 +473,9 @@ private TransactionOptions createTransactionOptions(final BsonDocument options)
468473
if (options.containsKey("readPreference")) {
469474
builder.readPreference(helper.getReadPreference(options));
470475
}
476+
if (options.containsKey("maxCommitTimeMS")) {
477+
builder.maxCommitTime(options.getNumber("maxCommitTimeMS").longValue(), MILLISECONDS);
478+
}
471479
return builder.build();
472480
}
473481

‎driver-core/src/main/com/mongodb/TransactionOptions.java‎

Lines changed: 54 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,11 @@
1919
import com.mongodb.annotations.Immutable;
2020
import com.mongodb.lang.Nullable;
2121

22+
import java.util.concurrent.TimeUnit;
23+
24+
import static com.mongodb.assertions.Assertions.isTrueArgument;
2225
import static com.mongodb.assertions.Assertions.notNull;
26+
import static java.util.concurrent.TimeUnit.MILLISECONDS;
2327

2428
/**
2529
* Options to apply to transactions. The default values for the options depend on context. For options specified per-transaction, the
@@ -36,6 +40,7 @@ public final class TransactionOptions {
3640
private final ReadConcern readConcern;
3741
private final WriteConcern writeConcern;
3842
private final ReadPreference readPreference;
43+
private final Long maxCommitTimeMS;
3944

4045
/**
4146
* Gets the read concern.
@@ -67,6 +72,24 @@ public ReadPreference getReadPreference() {
6772
return readPreference;
6873
}
6974

75+
/**
76+
* Gets the maximum amount of time to allow a single commitTransaction command to execute. The default is null, which places no
77+
* limit on the execution time.
78+
*
79+
* @param timeUnit the time unit to return the result in
80+
* @return the maximum execution time in the given time unit
81+
* @mongodb.server.release 4.2
82+
* @since 3.11
83+
*/
84+
@Nullable
85+
public Long getMaxCommitTime(final TimeUnit timeUnit) {
86+
notNull("timeUnit", timeUnit);
87+
if (maxCommitTimeMS == null) {
88+
return null;
89+
}
90+
return timeUnit.convert(maxCommitTimeMS, MILLISECONDS);
91+
}
92+
7093
/**
7194
* Gets an instance of a builder
7295
*
@@ -79,7 +102,7 @@ public static Builder builder() {
79102
/**
80103
* Merge the two provided transaction options, with the first taking precedence over the second.
81104
*
82-
* @param options the transaction options, which take precedence for any property that is non-null
105+
* @param options the transaction options, which take precedence for any property that is non-null
83106
* @param defaultOptions the default transaction options
84107
* @return the merged transaction options
85108
*/
@@ -93,6 +116,9 @@ public static TransactionOptions merge(final TransactionOptions options, final T
93116
? defaultOptions.getReadConcern() : options.getReadConcern())
94117
.readPreference(options.getReadPreference() == null
95118
? defaultOptions.getReadPreference() : options.getReadPreference())
119+
.maxCommitTime(options.getMaxCommitTime(MILLISECONDS) == null
120+
? defaultOptions.getMaxCommitTime(MILLISECONDS) : options.getMaxCommitTime(MILLISECONDS),
121+
MILLISECONDS)
96122
.build();
97123
}
98124

@@ -107,6 +133,9 @@ public boolean equals(final Object o) {
107133

108134
TransactionOptions that = (TransactionOptions) o;
109135

136+
if (maxCommitTimeMS != null ? !maxCommitTimeMS.equals(that.maxCommitTimeMS) : that.maxCommitTimeMS != null) {
137+
return false;
138+
}
110139
if (readConcern != null ? !readConcern.equals(that.readConcern) : that.readConcern != null) {
111140
return false;
112141
}
@@ -125,6 +154,7 @@ public int hashCode() {
125154
int result = readConcern != null ? readConcern.hashCode() : 0;
126155
result = 31 * result + (writeConcern != null ? writeConcern.hashCode() : 0);
127156
result = 31 * result + (readPreference != null ? readPreference.hashCode() : 0);
157+
result = 31 * result + (maxCommitTimeMS != null ? maxCommitTimeMS.hashCode() : 0);
128158
return result;
129159
}
130160

@@ -134,6 +164,7 @@ public String toString() {
134164
+ "readConcern=" + readConcern
135165
+ ", writeConcern=" + writeConcern
136166
+ ", readPreference=" + readPreference
167+
+ ", maxCommitTimeMS" + maxCommitTimeMS
137168
+ '}';
138169
}
139170

@@ -144,6 +175,7 @@ public static final class Builder {
144175
private ReadConcern readConcern;
145176
private WriteConcern writeConcern;
146177
private ReadPreference readPreference;
178+
private Long maxCommitTimeMS;
147179

148180
/**
149181
* Sets the read concern.
@@ -178,6 +210,26 @@ public Builder readPreference(@Nullable final ReadPreference readPreference) {
178210
return this;
179211
}
180212

213+
/**
214+
* Sets the maximum execution time on the server for the commitTransaction operation.
215+
*
216+
* @param maxCommitTime the max commit time, which must be either null or greater than zero, in the given time unit
217+
* @param timeUnit the time unit, which may not be null
218+
* @return this
219+
* @mongodb.server.release 4.2
220+
* @since 3.11
221+
*/
222+
public Builder maxCommitTime(@Nullable final Long maxCommitTime, final TimeUnit timeUnit) {
223+
if (maxCommitTime == null) {
224+
this.maxCommitTimeMS = null;
225+
} else {
226+
notNull("timeUnit", timeUnit);
227+
isTrueArgument("maxCommitTime > 0", maxCommitTime > 0);
228+
this.maxCommitTimeMS = MILLISECONDS.convert(maxCommitTime, timeUnit);
229+
}
230+
return this;
231+
}
232+
181233
/**
182234
* Build the transaction options instance.
183235
*
@@ -196,5 +248,6 @@ private TransactionOptions(final Builder builder) {
196248
readConcern = builder.readConcern;
197249
writeConcern = builder.writeConcern;
198250
readPreference = builder.readPreference;
251+
maxCommitTimeMS = builder.maxCommitTimeMS;
199252
}
200253
}

‎driver-core/src/main/com/mongodb/operation/AggregateOperationImpl.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,8 +51,8 @@
5151
import static com.mongodb.internal.async.ErrorHandlingResultCallback.errorHandlingCallback;
5252
import static com.mongodb.internal.operation.ServerVersionHelper.serverIsAtLeastVersionThreeDotSix;
5353
import static com.mongodb.operation.CommandOperationHelper.CommandCreator;
54-
import static com.mongodb.operation.CommandOperationHelper.executeCommandAsync;
5554
import static com.mongodb.operation.CommandOperationHelper.executeCommand;
55+
import static com.mongodb.operation.CommandOperationHelper.executeCommandAsync;
5656
import static com.mongodb.operation.OperationHelper.LOGGER;
5757
import static com.mongodb.operation.OperationHelper.cursorDocumentToQueryResult;
5858
import static com.mongodb.operation.OperationHelper.validateReadConcernAndCollation;
@@ -228,7 +228,8 @@ private BsonDocument getCommand(final ConnectionDescription description, final S
228228
appendReadConcernToCommand(sessionContext, commandDocument);
229229
commandDocument.put("pipeline", pipelineCreator.create(description, sessionContext));
230230
if (maxTimeMS > 0) {
231-
commandDocument.put("maxTimeMS", new BsonInt64(maxTimeMS));
231+
commandDocument.put("maxTimeMS", maxTimeMS > Integer.MAX_VALUE
232+
? new BsonInt64(maxTimeMS) : new BsonInt32((int) maxTimeMS));
232233
}
233234
if (!isInline(description)) {
234235
BsonDocument cursor = new BsonDocument();

‎driver-core/src/main/com/mongodb/operation/CommitTransactionOperation.java‎

Lines changed: 62 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818

1919
import com.mongodb.Function;
2020
import com.mongodb.MongoException;
21+
import com.mongodb.MongoExecutionTimeoutException;
2122
import com.mongodb.MongoNodeIsRecoveringException;
2223
import com.mongodb.MongoNotPrimaryException;
2324
import com.mongodb.MongoSocketException;
@@ -29,14 +30,20 @@
2930
import com.mongodb.binding.WriteBinding;
3031
import com.mongodb.connection.ConnectionDescription;
3132
import com.mongodb.connection.ServerDescription;
33+
import com.mongodb.lang.Nullable;
3234
import com.mongodb.operation.CommandOperationHelper.CommandCreator;
3335
import org.bson.BsonDocument;
36+
import org.bson.BsonInt32;
37+
import org.bson.BsonInt64;
3438

3539
import java.util.List;
3640
import java.util.concurrent.TimeUnit;
3741

3842
import static com.mongodb.MongoException.UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL;
43+
import static com.mongodb.assertions.Assertions.isTrueArgument;
44+
import static com.mongodb.assertions.Assertions.notNull;
3945
import static java.util.Arrays.asList;
46+
import static java.util.concurrent.TimeUnit.MILLISECONDS;
4047

4148
/**
4249
* An operation that commits a transaction.
@@ -47,6 +54,7 @@
4754
public class CommitTransactionOperation extends TransactionOperation {
4855
private final boolean alreadyCommitted;
4956
private BsonDocument recoveryToken;
57+
private Long maxCommitTimeMS;
5058

5159
/**
5260
* Construct an instance.
@@ -81,6 +89,44 @@ public CommitTransactionOperation recoveryToken(final BsonDocument recoveryToken
8189
return this;
8290
}
8391

92+
/**
93+
* Sets the maximum execution time on the server for the commitTransaction operation.
94+
*
95+
* @param maxCommitTime the max commit time, which must be either null or greater than zero, in the given time unit
96+
* @param timeUnit the time unit, which may not be null
97+
* @return this
98+
* @since 3.11
99+
* @mongodb.server.release 4.2
100+
*/
101+
public CommitTransactionOperation maxCommitTime(@Nullable final Long maxCommitTime, final TimeUnit timeUnit) {
102+
if (maxCommitTime == null) {
103+
this.maxCommitTimeMS = null;
104+
} else {
105+
notNull("timeUnit", timeUnit);
106+
isTrueArgument("maxCommitTime > 0", maxCommitTime > 0);
107+
this.maxCommitTimeMS = MILLISECONDS.convert(maxCommitTime, timeUnit);
108+
}
109+
return this;
110+
}
111+
112+
/**
113+
* Gets the maximum amount of time to allow a single commitTransaction command to execute. The default is 0, which places no limit on
114+
* the execution time.
115+
*
116+
* @param timeUnit the time unit to return the result in
117+
* @return the maximum execution time in the given time unit
118+
* @since 3.11
119+
* @mongodb.server.release 4.2
120+
*/
121+
@Nullable
122+
public Long getMaxCommitTime(final TimeUnit timeUnit) {
123+
notNull("timeUnit", timeUnit);
124+
if (maxCommitTimeMS == null) {
125+
return null;
126+
}
127+
return timeUnit.convert(maxCommitTimeMS, MILLISECONDS);
128+
}
129+
84130
@Override
85131
public Void execute(final WriteBinding binding) {
86132
try {
@@ -112,15 +158,16 @@ private void addErrorLabels(final MongoException e) {
112158

113159
private static final List<Integer> NON_RETRYABLE_WRITE_CONCERN_ERROR_CODES = asList(79, 100);
114160

115-
static boolean shouldAddUnknownTransactionCommitResultLabel(final Throwable t) {
161+
private static boolean shouldAddUnknownTransactionCommitResultLabel(final Throwable t) {
116162
if (!(t instanceof MongoException)) {
117163
return false;
118164
}
119165

120166
MongoException e = (MongoException) t;
121167

122168
if (e instanceof MongoSocketException || e instanceof MongoTimeoutException
123-
|| e instanceof MongoNotPrimaryException || e instanceof MongoNodeIsRecoveringException) {
169+
|| e instanceof MongoNotPrimaryException || e instanceof MongoNodeIsRecoveringException
170+
|| e instanceof MongoExecutionTimeoutException) {
124171
return true;
125172
}
126173

@@ -139,7 +186,19 @@ protected String getCommandName() {
139186

140187
@Override
141188
CommandCreator getCommandCreator() {
142-
final CommandCreator creator = super.getCommandCreator();
189+
final CommandCreator creator = new CommandCreator() {
190+
@Override
191+
public BsonDocument create(final ServerDescription serverDescription, final ConnectionDescription connectionDescription) {
192+
BsonDocument command = CommitTransactionOperation.super.getCommandCreator().create(serverDescription,
193+
connectionDescription);
194+
if (maxCommitTimeMS != null) {
195+
command.append("maxTimeMS",
196+
maxCommitTimeMS > Integer.MAX_VALUE
197+
? new BsonInt64(maxCommitTimeMS) : new BsonInt32(maxCommitTimeMS.intValue()));
198+
}
199+
return command;
200+
}
201+
};
143202
if (alreadyCommitted) {
144203
return new CommandCreator() {
145204
@Override

‎driver-core/src/test/functional/com/mongodb/client/CommandMonitoringTestHelper.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -267,6 +267,9 @@ private static CommandStartedEvent massageExpectedCommandStartedEvent(final Comm
267267
if (command.containsKey("autocommit") && command.isNull("autocommit")) {
268268
command.remove("autocommit");
269269
}
270+
if (command.containsKey("maxTimeMS") && command.isNull("maxTimeMS")) {
271+
command.remove("maxTimeMS");
272+
}
270273
if (command.containsKey("writeConcern") && command.isNull("writeConcern")) {
271274
command.remove("writeConcern");
272275
}

‎driver-core/src/test/functional/com/mongodb/operation/AggregateOperationSpecification.groovy‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -156,7 +156,7 @@ class AggregateOperationSpecification extends OperationFunctionalSpecification {
156156
.append('allowDiskUse', new BsonBoolean(true))
157157
.append('collation', defaultCollation.asDocument())
158158
.append('cursor', new BsonDocument('batchSize', new BsonInt32(10)))
159-
.append('maxTimeMS', new BsonInt64(10))
159+
.append('maxTimeMS', new BsonInt32(10))
160160

161161
then:
162162
testOperation(operation, [3, 4, 0], expectedCommand, async, helper.cursorResult)

‎driver-core/src/test/functional/com/mongodb/operation/CountOperationSpecification.groovy‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -306,7 +306,7 @@ class CountOperationSpecification extends OperationFunctionalSpecification {
306306
new BsonDocument('$skip', new BsonInt64(30)),
307307
new BsonDocument('$limit', new BsonInt64(20)),
308308
pipeline.last()]))
309-
.append('maxTimeMS', new BsonInt64(10))
309+
.append('maxTimeMS', new BsonInt32(10))
310310
.append('collation', defaultCollation.asDocument())
311311
.append('hint', hint)
312312

0 commit comments

Comments
 (0)