翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# Amazon DocumentDB でのトランザクション
Amazon DocumentDB (MongoDB 互換) は、ACID トランザクションを含む MongoDB 4.0 の互換性をサポートしています。複数のドキュメント、ステートメント、コレクション、およびデータベース間でトランザクションを実行できます。トランザクションは、Amazon DocumentDB クラスター内の 1 つまたは複数のドキュメントに対し、原始的、一貫性、分離、耐久性（ACID）オペレーションを実行できるようにすることで、アプリケーション開発を簡素化します。トランザクションの一般的なユースケースには、財務処理、注文の履行と管理、マルチプレイヤーゲームの構築などがあります。
トランザクションを有効にするために、追加料金はかかりません。支払いは、トランザクションの一部として消費した読み書きの IO に対してのみ支払います。
**Topics**
+ [要件](#transactions-requirements)
+ [ベストプラクティス](#transactions-best-practices)
+ [制限](#transactions-limitations)
+ [モニタリングと診断](#transactions-monitoring)
+ [トランザクション分離レベル](#transactions-isolation-level)
+ [ユースケース](#transactions-usecases)
+ [サポートされているコマンド](#transactions-supported-commands)
+ [サポート対象外の機能](#transactions-unsupported-commands)
+ [セッション](#transactions-sessions)
+ [トランザクションエラー](#transactions-errors)
## 要件
トランザクション機能を使用するには、次の要件を満たしている必要があります。
+ Amazon DocumentDB 4.0 エンジンを使用する必要があります。
+ MongoDB 4.0 以降と互換性のあるドライバを使用する必要があります。
## ベストプラクティス
Amazon DocumentDB でトランザクションを最も多く利用できるようにするためのベストプラクティスをいくつか紹介します。
+ トランザクションの完了後は、必ずトランザクションをコミットまたは中断してください。トランザクションが不完全な状態のままにすると、データベースリソースが結びつき、書き込みの競合が発生する可能性があります。
+ トランザクションは、必要なコマンド数を最小限に抑えることをお勧めします。複数のステートメントを持つトランザクションを複数の小さなトランザクションに分割できる場合は、タイムアウトの可能性を減らすためにこれを行うことをお勧めします。長時間実行される読み取りではなく、短いトランザクションを作成することを常に目指してください。
## 制限
+ Amazon DocumentDB は、トランザクション内のカーソルをサポートしていません。
+ Amazon DocumentDB はトランザクションで新しいコレクションを作成できず、存在しないコレクションに対してクエリ/更新することはできません。
+ ドキュメントレベルの書き込みロックには 1 分間のタイムアウトが適用されますが、これはユーザーによって設定できません。
+ Amazon DocumentDB では、再試行可能な書き込み、再試行可能なコミット、および再試行可能な中止コマンドはサポートされていません。例外: mongo シェル を使用している場合は、どのコード文字列にも `retryWrites=false` コマンドを含めないでください。デフォルトでは、再試行可能な書き込みは無効になっています。`retryWrites=false` を含めると、通常の読み取りコマンドでは失敗する可能性があります。
+ 各 Amazon DocumentDB インスタンスには、一度にインスタンスでオープンされる同時トランザクションの数に上限があります。制限事項については、「[インスタンス制限](limits.md#limits.instance)」を参照してください。
+ 特定のトランザクションでは、トランザクションログのサイズは 32 MB 未満である必要があります。
+ Amazon DocumentDB はトランザクション内で `count()` をサポートしていますが、すべてのドライバがこの機能をサポートしているわけではありません。別の方法として、`countDocuments()` API を使用して、count クエリをクライアント側の集計クエリに変換します。
+ トランザクションの実行制限は 1 分で、セッションには 30 分のタイムアウトがあります。トランザクションがタイムアウトすると、トランザクションは中止され、既存のトランザクションのセッション内で後続のコマンドが発行されると、次のエラーが表示されます。
```
WriteCommandError({
"ok" : 0,
"operationTime" : Timestamp(1603491424, 627726),
"code" : 251,
"errmsg" : "Given transaction number 0 does not match any in-progress transactions."
}
```
## モニタリングと診断
Amazon DocumentDB 4.0 でのトランザクションのサポートにより、トランザクションのモニタリングに役立つ追加の CloudWatch メトリックスが追加されました。
新しい CloudWatch メトリクス
+ `DatabaseTransactions`: 1 分間に実行された未処理のトランザクションの数。
+ `DatabaseTransactionsAborted`: 1 分間に中断された未処理のトランザクションの数。
+ `DatabaseTransactionsMax`: 1 分間にインスタンスで開かれるトランザクションの最大数。
+ `TransactionsAborted`: 1 分間にインスタンスで中断されたトランザクションの数。
+ `TransactionsCommitted`: 1 分間にインスタンスでコミットされたトランザクションの数。
+ `TransactionsOpen`: 1 分間で実行されるインスタンスで開かれるトランザクションの数。
+ `TransactionsOpenMax`: 1 分間にインスタンスで開いているトランザクションの最大数。
+ `TransactionsStarted`: 1 分間にインスタンスで開始されたトランザクションの数。
**注記**
Amazon DocumentDB の CloudWatch メトリックスの詳細については、[Amazon DocumentDB と CloudWatch のモニタリング](cloud_watch.md)。
さらに、新しいフィールドが `currentOp` `lsid`、`transactionThreadId`、`idle transaction` のための新しい状態、および `serverStatus` トランザクション: `currentActive`、`currentInactive`、 `currentOpen`、`totalAborted`、`totalCommitted`、および `totalStarted` に追加されました。
## トランザクション分離レベル
トランザクションを開始するとき、次の例に示されるように `readConcern` と `writeConcern` の両方を指定することができます。
`mySession.startTransaction({readConcern: {level: 'snapshot'}, writeConcern: {w: 'majority'}});`
`readConcern` を使用する場合、Amazon DocumentDB はデフォルトでスナップショットアイソレーションをサポートしています。`readConcern` に local、available、majority が指定された場合、Amazon DocumentDB は `readConcern` レベルをスナプショットにアップグレードします。Amazon DocumentDB は線形化可能な `readConcern` をサポートしていません。このような読み取り懸念を指定すると、エラーが発生します。
`writeConcern` を使用する場合、Amazon DocumentDB はデフォルトで多数数をサポートしており、3 つの AZ にわたってデータの 4 つのコピーが永続化されると、書き込みクォーラムが達成されます。低い `writeConcern` が指定されている場合、Amazon DocumentDB は `writeConcern` を過半数にアップグレードします。さらに、Amazon DocumentDB の書き込みはすべてジャーナル処理され、ジャーナリングを無効にすることはできません。
## ユースケース
このセクションでは、トランザクションの 2 つのユースケース、マルチステートメントとマルチコレクションについて説明します。
### マルチステートメントトランザクション
Amazon DocumentDB トランザクションはマルチステートメントです。つまり、明示的なコミットまたはロールバックを使用して、複数のステートメントにまたがるトランザクションを記述できます。単一のアトミック操作として、`insert`、`update`、`delete`、および `findAndModify` のアクションをグループ分けできます。
マルチステートメントトランザクションの一般的なユースケースは、デビットクレジットトランザクションです。例えば、友達に服のお金を借りたとします。アカウントから 500 ドルのデビット (引き出し) を実行し、友達のアカウントに 500 ドル (入金) をクレジットする必要があります。その操作を実行するには、原子性を保証するために、単一のトランザクション内で債務操作と与信操作の両方を実行します。これにより、500 ドルがアカウントから引き落とされ、友達のアカウントに入金されないシナリオが防止されます。このユースケースは次のようになります。
```
// *** Transfer $500 from Alice to Bob inside a transaction: Success Scenario***
// Setup bank account for Alice and Bob. Each have $1000 in their account
var databaseName = "bank";
var collectionName = "account";
var amountToTransfer = 500;
var session = db.getMongo().startSession({causalConsistency: false});
var bankDB = session.getDatabase(databaseName);
var accountColl = bankDB[collectionName];
accountColl.drop();
accountColl.insert({name: "Alice", balance: 1000});
accountColl.insert({name: "Bob", balance: 1000});
session.startTransaction();
// deduct $500 from Alice's account
var aliceBalance = accountColl.find({"name": "Alice"}).next().balance;
var newAliceBalance = aliceBalance - amountToTransfer;
accountColl.update({"name": "Alice"},{"$set": {"balance": newAliceBalance}});
var findAliceBalance = accountColl.find({"name": "Alice"}).next().balance;
// add $500 to Bob's account
var bobBalance = accountColl.find({"name": "Bob"}).next().balance;
var newBobBalance = bobBalance + amountToTransfer;
accountColl.update({"name": "Bob"},{"$set": {"balance": newBobBalance}});
var findBobBalance = accountColl.find({"name": "Bob"}).next().balance;
session.commitTransaction();
accountColl.find();
// *** Transfer $500 from Alice to Bob inside a transaction: Failure Scenario***
// Setup bank account for Alice and Bob. Each have $1000 in their account
var databaseName = "bank";
var collectionName = "account";
var amountToTransfer = 500;
var session = db.getMongo().startSession({causalConsistency: false});
var bankDB = session.getDatabase(databaseName);
var accountColl = bankDB[collectionName];
accountColl.drop();
accountColl.insert({name: "Alice", balance: 1000});
accountColl.insert({name: "Bob", balance: 1000});
session.startTransaction();
// deduct $500 from Alice's account
var aliceBalance = accountColl.find({"name": "Alice"}).next().balance;
var newAliceBalance = aliceBalance - amountToTransfer;
accountColl.update({"name": "Alice"},{"$set": {"balance": newAliceBalance}});
var findAliceBalance = accountColl.find({"name": "Alice"}).next().balance;
session.abortTransaction();
```
### マルチコレクショントランザクション
また、トランザクションはマルチコレクションです。つまり、単一のトランザクション内および複数のコレクションにわたって複数の操作を実行するために使用できます。これにより、データの整合性が保たれ、データの整合性が維持されます。コマンドを単一の `<>` としてコミットする場合、トランザクションはすべてまたは何も実行されません。その場合、トランザクションはすべて成功するか、すべて失敗します。
以下は、マルチコレクショントランザクションの例で、複数ステートメントトランザクションの例と同じシナリオとデータを使用します。
```
// *** Transfer $500 from Alice to Bob inside a transaction: Success Scenario***
// Setup bank account for Alice and Bob. Each have $1000 in their account
var amountToTransfer = 500;
var collectionName = "account";
var session = db.getMongo().startSession({causalConsistency: false});
var accountCollInBankA = session.getDatabase("bankA")[collectionName];
var accountCollInBankB = session.getDatabase("bankB")[collectionName];
accountCollInBankA.drop();
accountCollInBankB.drop();
accountCollInBankA.insert({name: "Alice", balance: 1000});
accountCollInBankB.insert({name: "Bob", balance: 1000});
session.startTransaction();
// deduct $500 from Alice's account
var aliceBalance = accountCollInBankA.find({"name": "Alice"}).next().balance;
var newAliceBalance = aliceBalance - amountToTransfer;
accountCollInBankA.update({"name": "Alice"},{"$set": {"balance": newAliceBalance}});
var findAliceBalance = accountCollInBankA.find({"name": "Alice"}).next().balance;
// add $500 to Bob's account
var bobBalance = accountCollInBankB.find({"name": "Bob"}).next().balance;
var newBobBalance = bobBalance + amountToTransfer;
accountCollInBankB.update({"name": "Bob"},{"$set": {"balance": newBobBalance}});
var findBobBalance = accountCollInBankB.find({"name": "Bob"}).next().balance;
session.commitTransaction();
accountCollInBankA.find(); // Alice holds $500 in bankA
accountCollInBankB.find(); // Bob holds $1500 in bankB
// *** Transfer $500 from Alice to Bob inside a transaction: Failure Scenario***
// Setup bank account for Alice and Bob. Each have $1000 in their account
var collectionName = "account";
var amountToTransfer = 500;
var session = db.getMongo().startSession({causalConsistency: false});
var accountCollInBankA = session.getDatabase("bankA")[collectionName];
var accountCollInBankB = session.getDatabase("bankB")[collectionName];
accountCollInBankA.drop();
accountCollInBankB.drop();
accountCollInBankA.insert({name: "Alice", balance: 1000});
accountCollInBankB.insert({name: "Bob", balance: 1000});
session.startTransaction();
// deduct $500 from Alice's account
var aliceBalance = accountCollInBankA.find({"name": "Alice"}).next().balance;
var newAliceBalance = aliceBalance - amountToTransfer;
accountCollInBankA.update({"name": "Alice"},{"$set": {"balance": newAliceBalance}});
var findAliceBalance = accountCollInBankA.find({"name": "Alice"}).next().balance;
// add $500 to Bob's account
var bobBalance = accountCollInBankB.find({"name": "Bob"}).next().balance;
var newBobBalance = bobBalance + amountToTransfer;
accountCollInBankB.update({"name": "Bob"},{"$set": {"balance": newBobBalance}});
var findBobBalance = accountCollInBankB.find({"name": "Bob"}).next().balance;
session.abortTransaction();
accountCollInBankA.find(); // Alice holds $1000 in bankA
accountCollInBankB.find(); // Bob holds $1000 in bankB
```
### コールバック API のトランザクション API の例
コールバック API は 4.2 以降のドライバでのみ使用できます。
------
#### [ Javascript ]
次のコードは、Javascript で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// *** Transfer $500 from Alice to Bob inside a transaction: Success ***
// Setup bank account for Alice and Bob. Each have $1000 in their account
var databaseName = "bank";
var collectionName = "account";
var amountToTransfer = 500;
var session = db.getMongo().startSession({causalConsistency: false});
var bankDB = session.getDatabase(databaseName);
var accountColl = bankDB[collectionName];
accountColl.drop();
accountColl.insert({name: "Alice", balance: 1000});
accountColl.insert({name: "Bob", balance: 1000});
session.startTransaction();
// deduct $500 from Alice's account
var aliceBalance = accountColl.find({"name": "Alice"}).next().balance;
assert(aliceBalance >= amountToTransfer);
var newAliceBalance = aliceBalance - amountToTransfer;
accountColl.update({"name": "Alice"},{"$set": {"balance": newAliceBalance}});
var findAliceBalance = accountColl.find({"name": "Alice"}).next().balance;
assert.eq(newAliceBalance, findAliceBalance);
// add $500 to Bob's account
var bobBalance = accountColl.find({"name": "Bob"}).next().balance;
var newBobBalance = bobBalance + amountToTransfer;
accountColl.update({"name": "Bob"},{"$set": {"balance": newBobBalance}});
var findBobBalance = accountColl.find({"name": "Bob"}).next().balance;
assert.eq(newBobBalance, findBobBalance);
session.commitTransaction();
accountColl.find();
```
------
#### [ Node.js ]
次のコードは、Node.js で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// Node.js callback API:
const bankDB = await mongoclient.db("bank");
var accountColl = await bankDB.createCollection("account");
var amountToTransfer = 500;
const session = mongoclient.startSession({causalConsistency: false});
await accountColl.drop();
await accountColl.insertOne({name: "Alice", balance: 1000}, { session });
await accountColl.insertOne({name: "Bob", balance: 1000}, { session });
const transactionOptions = {
readConcern: { level: 'snapshot' },
writeConcern: { w: 'majority' }
};
// deduct $500 from Alice's account
var aliceBalance = await accountColl.findOne({name: "Alice"}, {session});
assert(aliceBalance.balance >= amountToTransfer);
var newAliceBalance = aliceBalance - amountToTransfer;
session.startTransaction(transactionOptions);
await accountColl.updateOne({name: "Alice"}, {$set: {balance: newAliceBalance}}, {session });
await session.commitTransaction();
aliceBalance = await accountColl.findOne({name: "Alice"}, {session});
assert(newAliceBalance == aliceBalance.balance);
// add $500 to Bob's account
var bobBalance = await accountColl.findOne({name: "Bob"}, {session});
var newBobBalance = bobBalance.balance + amountToTransfer;
session.startTransaction(transactionOptions);
await accountColl.updateOne({name: "Bob"}, {$set: {balance: newBobBalance}}, {session });
await session.commitTransaction();
bobBalance = await accountColl.findOne({name: "Bob"}, {session});
assert(newBobBalance == bobBalance.balance);
```
------
#### [ C\$1 ]
次のコードは、C\$1 で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// C# Callback API
var dbName = "bank";
var collName = "account";
var amountToTransfer = 500;
using (var session = client.StartSession(new ClientSessionOptions{CausalConsistency = false}))
{
var bankDB = client.GetDatabase(dbName);
var accountColl = bankDB.GetCollection(collName);
bankDB.DropCollection(collName);
accountColl.InsertOne(session, new BsonDocument { {"name", "Alice"}, {"balance", 1000 } });
accountColl.InsertOne(session, new BsonDocument { {"name", "Bob"}, {"balance", 1000 } });
// start transaction
var transactionOptions = new TransactionOptions(
readConcern: ReadConcern.Snapshot,
writeConcern: WriteConcern.WMajority);
var result = session.WithTransaction(
(sess, cancellationtoken) =>
{
// deduct $500 from Alice's account
var aliceBalance = accountColl.Find(sess, Builders.Filter.Eq("name", "Alice")).FirstOrDefault().GetValue("balance");
Debug.Assert(aliceBalance >= amountToTransfer);
var newAliceBalance = aliceBalance.AsInt32 - amountToTransfer;
accountColl.UpdateOne(sess, Builders.Filter.Eq("name", "Alice"),
Builders.Update.Set("balance", newAliceBalance));
aliceBalance = accountColl.Find(sess, Builders.Filter.Eq("name", "Alice")).FirstOrDefault().GetValue("balance");
Debug.Assert(aliceBalance == newAliceBalance);
// add $500 from Bob's account
var bobBalance = accountColl.Find(sess, Builders.Filter.Eq("name", "Bob")).FirstOrDefault().GetValue("balance");
var newBobBalance = bobBalance.AsInt32 + amountToTransfer;
accountColl.UpdateOne(sess, Builders.Filter.Eq("name", "Bob"),
Builders.Update.Set("balance", newBobBalance));
bobBalance = accountColl.Find(sess, Builders.Filter.Eq("name", "Bob")).FirstOrDefault().GetValue("balance");
Debug.Assert(bobBalance == newBobBalance);
return "Transaction committed";
}, transactionOptions);
// check values outside of transaction
var aliceNewBalance = accountColl.Find(Builders.Filter.Eq("name", "Alice")).FirstOrDefault().GetValue("balance");
var bobNewBalance = accountColl.Find(Builders.Filter.Eq("name", "Bob")).FirstOrDefault().GetValue("balance");
Debug.Assert(aliceNewBalance == 500);
Debug.Assert(bobNewBalance == 1500);
}
```
------
#### [ Ruby ]
次のコードは、Ruby で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// Ruby Callback API
dbName = "bank"
collName = "account"
amountToTransfer = 500
session = client.start_session(:causal_consistency=> false)
bankDB = Mongo::Database.new(client, dbName)
accountColl = bankDB[collName]
accountColl.drop()
accountColl.insert_one({"name"=>"Alice", "balance"=>1000})
accountColl.insert_one({"name"=>"Bob", "balance"=>1000})
# start transaction
session.with_transaction(read_concern: {level: :snapshot}, write_concern: {w: :majority}) do
# deduct $500 from Alice's account
aliceBalance = accountColl.find({"name"=>"Alice"}, :session=> session).first['balance']
assert aliceBalance >= amountToTransfer
newAliceBalance = aliceBalance - amountToTransfer
accountColl.update_one({"name"=>"Alice"}, { "$set" => {"balance"=>newAliceBalance} }, :session=> session)
aliceBalance = accountColl.find({"name"=>>"Alice"}, :session=> session).first['balance']
assert_equal(newAliceBalance, aliceBalance)
# add $500 from Bob's account
bobBalance = accountColl.find({"name"=>"Bob"}, :session=> session).first['balance']
newBobBalance = bobBalance + amountToTransfer
accountColl.update_one({"name"=>"Bob"}, { "$set" => {"balance"=>newBobBalance} }, :session=> session)
bobBalance = accountColl.find({"name"=>"Bob"}, :session=> session).first['balance']
assert_equal(newBobBalance, bobBalance)
end
# check results outside of transaction
aliceBalance = accountColl.find({"name"=>"Alice"}).first['balance']
bobBalance = accountColl.find({"name"=>"Bob"}).first['balance']
assert_equal(aliceBalance, 500)
assert_equal(bobBalance, 1500)
session.end_session
```
------
#### [ Go ]
次のコードは、Go で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// Go - Callback API
type Account struct {
Name string
Balance int
}
ctx := context.TODO()
dbName := "bank"
collName := "account"
amountToTransfer := 500
session, err := client.StartSession(options.Session().SetCausalConsistency(false))
assert.NilError(t, err)
defer session.EndSession(ctx)
bankDB := client.Database(dbName)
accountColl := bankDB.Collection(collName)
accountColl.Drop(ctx)
_, err = accountColl.InsertOne(ctx, bson.M{"name" : "Alice", "balance":1000})
_, err = accountColl.InsertOne(ctx, bson.M{"name" : "Bob", "balance":1000})
transactionOptions := options.Transaction().SetReadConcern(readconcern.Snapshot()).
SetWriteConcern(writeconcern.New(writeconcern.WMajority()))
_, err = session.WithTransaction(ctx, func(sessionCtx mongo.SessionContext) (interface{}, error) {
var result Account
// deduct $500 from Alice's account
err = accountColl.FindOne(sessionCtx, bson.M{"name": "Alice"}).Decode(&result)
aliceBalance := result.Balance
newAliceBalance := aliceBalance - amountToTransfer
_, err = accountColl.UpdateOne(sessionCtx, bson.M{"name": "Alice"}, bson.M{"$set": bson.M{"balance": newAliceBalance}})
err = accountColl.FindOne(sessionCtx, bson.M{"name": "Alice"}).Decode(&result)
aliceBalance = result.Balance
assert.Equal(t, aliceBalance, newAliceBalance)
// add $500 to Bob's account
err = accountColl.FindOne(sessionCtx, bson.M{"name": "Bob"}).Decode(&result)
bobBalance := result.Balance
newBobBalance := bobBalance + amountToTransfer
_, err = accountColl.UpdateOne(sessionCtx, bson.M{"name": "Bob"}, bson.M{"$set": bson.M{"balance": newBobBalance}})
err = accountColl.FindOne(sessionCtx, bson.M{"name": "Bob"}).Decode(&result)
bobBalance = result.Balance
assert.Equal(t, bobBalance, newBobBalance)
if err != nil {
return nil, err
}
return "transaction committed", err
}, transactionOptions)
// check results outside of transaction
var result Account
err = accountColl.FindOne(ctx, bson.M{"name": "Alice"}).Decode(&result)
aliceNewBalance := result.Balance
err = accountColl.FindOne(ctx, bson.M{"name": "Bob"}).Decode(&result)
bobNewBalance := result.Balance
assert.Equal(t, aliceNewBalance, 500)
assert.Equal(t, bobNewBalance, 1500)
```
------
#### [ Java ]
次のコードは、Java で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// Java (sync) - Callback API
MongoDatabase bankDB = mongoClient.getDatabase("bank");
MongoCollection accountColl = bankDB.getCollection("account");
accountColl.drop();
int amountToTransfer = 500;
// add sample data
accountColl.insertOne(new Document("name", "Alice").append("balance", 1000));
accountColl.insertOne(new Document("name", "Bob").append("balance", 1000));
TransactionOptions txnOptions = TransactionOptions.builder()
.readConcern(ReadConcern.SNAPSHOT)
.writeConcern(WriteConcern.MAJORITY)
.build();
ClientSessionOptions sessionOptions = ClientSessionOptions.builder().causallyConsistent(false).build();
try ( ClientSession clientSession = mongoClient.startSession(sessionOptions) ) {
clientSession.withTransaction(new TransactionBody() {
@Override
public Void execute() {
// deduct $500 from Alice's account
List documentList = new ArrayList<>();
accountColl.find(clientSession, new Document("name", "Alice")).into(documentList);
int aliceBalance = (int) documentList.get(0).get("balance");
int newAliceBalance = aliceBalance - amountToTransfer;
accountColl.updateOne(clientSession, new Document("name", "Alice"), new Document("$set", new Document("balance", newAliceBalance)));
// check Alice's new balance
documentList = new ArrayList<>();
accountColl.find(clientSession, new Document("name", "Alice")).into(documentList);
int updatedBalance = (int) documentList.get(0).get("balance");
Assert.assertEquals(updatedBalance, newAliceBalance);
// add $500 to Bob's account
documentList = new ArrayList<>();
accountColl.find(clientSession, new Document("name", "Bob")).into(documentList);
int bobBalance = (int) documentList.get(0).get("balance");
int newBobBalance = bobBalance + amountToTransfer;
accountColl.updateOne(clientSession, new Document("name", "Bob"), new Document("$set", new Document("balance", newBobBalance)));
// check Bob's new balance
documentList = new ArrayList<>();
accountColl.find(clientSession, new Document("name", "Bob")).into(documentList);
updatedBalance = (int) documentList.get(0).get("balance");
Assert.assertEquals(updatedBalance, newBobBalance);
return null;
}
}, txnOptions);
}
```
------
#### [ C ]
次のコードは、C で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// Sample Code for C with Callback
#include
#include
#include
#include
#include
typedef struct {
int64_t balance;
bson_t *account;
bson_t *opts;
mongoc_collection_t *collection;
} ctx_t;
bool callback_session (mongoc_client_session_t *session, void *ctx, bson_t **reply, bson_error_t *error)
{
bool r = true;
ctx_t *data = (ctx_t *) ctx;
bson_t local_reply;
bson_t *selector = data->account;
bson_t *update = BCON_NEW ("$set", "{", "balance", BCON_INT64 (data->balance), "}");
mongoc_collection_update_one (data->collection, selector, update, data->opts, &local_reply, error);
*reply = bson_copy (&local_reply);
bson_destroy (&local_reply);
bson_destroy (update);
return r;
}
void test_callback_money_transfer(mongoc_client_t* client, mongoc_collection_t* collection, int amount_to_transfer){
bson_t reply;
bool r = true;
const bson_t *doc;
bson_iter_t iter;
ctx_t alice_ctx;
ctx_t bob_ctx;
bson_error_t error;
// find query
bson_t *alice_query = bson_new ();
BSON_APPEND_UTF8(alice_query, "name", "Alice");
bson_t *bob_query = bson_new ();
BSON_APPEND_UTF8(bob_query, "name", "Bob");
// create session
// set causal consistency to false
mongoc_session_opt_t *session_opts = mongoc_session_opts_new ();
mongoc_session_opts_set_causal_consistency (session_opts, false);
// start the session
mongoc_client_session_t *client_session = mongoc_client_start_session (client, session_opts, &error);
// add session to options
bson_t *opts = bson_new();
mongoc_client_session_append (client_session, opts, &error);
// deduct 500 from Alice
// find account balance of Alice
mongoc_cursor_t *cursor = mongoc_collection_find_with_opts (collection, alice_query, NULL, NULL);
mongoc_cursor_next (cursor, &doc);
bson_iter_init (&iter, doc);
bson_iter_find (&iter, "balance");
int64_t alice_balance = (bson_iter_value (&iter))->value.v_int64;
assert(alice_balance >= amount_to_transfer);
int64_t new_alice_balance = alice_balance - amount_to_transfer;
// set variables which will be used by callback function
alice_ctx.collection = collection;
alice_ctx.opts = opts;
alice_ctx.balance = new_alice_balance;
alice_ctx.account = alice_query;
// callback
r = mongoc_client_session_with_transaction (client_session, &callback_session, NULL, &alice_ctx, &reply, &error);
assert(r);
// find account balance of Alice after transaction
cursor = mongoc_collection_find_with_opts (collection, alice_query, NULL, NULL);
mongoc_cursor_next (cursor, &doc);
bson_iter_init (&iter, doc);
bson_iter_find (&iter, "balance");
alice_balance = (bson_iter_value (&iter))->value.v_int64;
assert(alice_balance == new_alice_balance);
assert(alice_balance == 500);
// add 500 to bob's balance
// find account balance of Bob
cursor = mongoc_collection_find_with_opts (collection, bob_query, NULL, NULL);
mongoc_cursor_next (cursor, &doc);
bson_iter_init (&iter, doc);
bson_iter_find (&iter, "balance");
int64_t bob_balance = (bson_iter_value (&iter))->value.v_int64;
int64_t new_bob_balance = bob_balance + amount_to_transfer;
bob_ctx.collection = collection;
bob_ctx.opts = opts;
bob_ctx.balance = new_bob_balance;
bob_ctx.account = bob_query;
// set read & write concern
mongoc_read_concern_t *read_concern = mongoc_read_concern_new ();
mongoc_write_concern_t *write_concern = mongoc_write_concern_new ();
mongoc_transaction_opt_t *txn_opts = mongoc_transaction_opts_new ();
mongoc_write_concern_set_w(write_concern, MONGOC_WRITE_CONCERN_W_MAJORITY);
mongoc_read_concern_set_level(read_concern, MONGOC_READ_CONCERN_LEVEL_SNAPSHOT);
mongoc_transaction_opts_set_write_concern (txn_opts, write_concern);
mongoc_transaction_opts_set_read_concern (txn_opts, read_concern);
// callback
r = mongoc_client_session_with_transaction (client_session, &callback_session, txn_opts, &bob_ctx, &reply, &error);
assert(r);
// find account balance of Bob after transaction
cursor = mongoc_collection_find_with_opts (collection, bob_query, NULL, NULL);
mongoc_cursor_next (cursor, &doc);
bson_iter_init (&iter, doc);
bson_iter_find (&iter, "balance");
bob_balance = (bson_iter_value (&iter))->value.v_int64;
assert(bob_balance == new_bob_balance);
assert(bob_balance == 1500);
// cleanup
bson_destroy(alice_query);
bson_destroy(bob_query);
mongoc_client_session_destroy(client_session);
bson_destroy(opts);
mongoc_transaction_opts_destroy(txn_opts);
mongoc_read_concern_destroy(read_concern);
mongoc_write_concern_destroy(write_concern);
mongoc_cursor_destroy(cursor);
bson_destroy(doc);
}
int main(int argc, char* argv[]) {
mongoc_init ();
mongoc_client_t* client = mongoc_client_new ();
bson_error_t error;
// connect to bank db
mongoc_database_t *database = mongoc_client_get_database (client, "bank");
// access account collection
mongoc_collection_t* collection = mongoc_client_get_collection(client, "bank", "account");
// set amount to transfer
int64_t amount_to_transfer = 500;
// delete the collection if already existing
mongoc_collection_drop(collection, &error);
// open Alice account
bson_t *alice_account = bson_new ();
BSON_APPEND_UTF8(alice_account, "name", "Alice");
BSON_APPEND_INT64(alice_account, "balance", 1000);
// open Bob account
bson_t *bob_account = bson_new ();
BSON_APPEND_UTF8(bob_account, "name", "Bob");
BSON_APPEND_INT64(bob_account, "balance", 1000);
bool r = true;
r = mongoc_collection_insert_one(collection, alice_account, NULL, NULL, &error);
if (!r) {printf("Error encountered:%s", error.message);}
r = mongoc_collection_insert_one(collection, bob_account, NULL, NULL, &error);
if (!r) {printf("Error encountered:%s", error.message);}
test_callback_money_transfer(client, collection, amount_to_transfer);
}
```
------
#### [ Python ]
次のコードは、Python で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// Sample Python code with callback api
import pymongo
def callback(session, balance, query):
collection.update_one(query, {'$set': {"balance": balance}}, session=session)
client = pymongo.MongoClient()
rc_snapshot = pymongo.read_concern.ReadConcern('snapshot')
wc_majority = pymongo.write_concern.WriteConcern('majority')
# To start, drop and create an account collection and insert balances for both Alice and Bob
collection = client.get_database("bank").get_collection("account")
collection.drop()
collection.insert_one({"_id": 1, "name": "Alice", "balance": 1000})
collection.insert_one({"_id": 2, "name": "Bob", "balance": 1000})
amount_to_transfer = 500
# deduct 500 from Alice's account
alice_balance = collection.find_one({"name": "Alice"}).get("balance")
assert alice_balance >= amount_to_transfer
new_alice_balance = alice_balance - amount_to_transfer
with client.start_session({'causalConsistency':False}) as session:
session.with_transaction(lambda s: callback(s, new_alice_balance, {"name": "Alice"}), read_concern=rc_snapshot, write_concern=wc_majority)
updated_alice_balance = collection.find_one({"name": "Alice"}).get("balance")
assert updated_alice_balance == new_alice_balance
# add 500 to Bob's account
bob_balance = collection.find_one({"name": "Bob"}).get("balance")
assert bob_balance >= amount_to_transfer
new_bob_balance = bob_balance + amount_to_transfer
with client.start_session({'causalConsistency':False}) as session:
session.with_transaction(lambda s: callback(s, new_bob_balance, {"name": "Bob"}), read_concern=rc_snapshot, write_concern=wc_majority)
updated_bob_balance = collection.find_one({"name": "Bob"}).get("balance")
assert updated_bob_balance == new_bob_balance
```
------
### コア API のトランザクション API の例
------
#### [ Javascript ]
次のコードは、Javascript で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// *** Transfer $500 from Alice to Bob inside a transaction: Success ***
// Setup bank account for Alice and Bob. Each have $1000 in their account
var databaseName = "bank";
var collectionName = "account";
var amountToTransfer = 500;
var session = db.getMongo().startSession({causalConsistency: false});
var bankDB = session.getDatabase(databaseName);
var accountColl = bankDB[collectionName];
accountColl.drop();
accountColl.insert({name: "Alice", balance: 1000});
accountColl.insert({name: "Bob", balance: 1000});
session.startTransaction();
// deduct $500 from Alice's account
var aliceBalance = accountColl.find({"name": "Alice"}).next().balance;
assert(aliceBalance >= amountToTransfer);
var newAliceBalance = aliceBalance - amountToTransfer;
accountColl.update({"name": "Alice"},{"$set": {"balance": newAliceBalance}});
var findAliceBalance = accountColl.find({"name": "Alice"}).next().balance;
assert.eq(newAliceBalance, findAliceBalance);
// add $500 to Bob's account
var bobBalance = accountColl.find({"name": "Bob"}).next().balance;
var newBobBalance = bobBalance + amountToTransfer;
accountColl.update({"name": "Bob"},{"$set": {"balance": newBobBalance}});
var findBobBalance = accountColl.find({"name": "Bob"}).next().balance;
assert.eq(newBobBalance, findBobBalance);
session.commitTransaction();
accountColl.find();
```
------
#### [ C\$1 ]
次のコードは、C\$1 で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// C# Core API
public void TransferMoneyWithRetry(IMongoCollection accountColl, IClientSessionHandle session)
{
var amountToTransfer = 500;
// start transaction
var transactionOptions = new TransactionOptions(
readConcern: ReadConcern.Snapshot,
writeConcern: WriteConcern.WMajority);
session.StartTransaction(transactionOptions);
try
{
// deduct $500 from Alice's account
var aliceBalance = accountColl.Find(session, Builders.Filter.Eq("name", "Alice")).FirstOrDefault().GetValue("balance");
Debug.Assert(aliceBalance >= amountToTransfer);
var newAliceBalance = aliceBalance.AsInt32 - amountToTransfer;
accountColl.UpdateOne(session, Builders.Filter.Eq("name", "Alice"),
Builders.Update.Set("balance", newAliceBalance));
aliceBalance = accountColl.Find(session, Builders.Filter.Eq("name", "Alice")).FirstOrDefault().GetValue("balance");
Debug.Assert(aliceBalance == newAliceBalance);
// add $500 from Bob's account
var bobBalance = accountColl.Find(session, Builders.Filter.Eq("name", "Bob")).FirstOrDefault().GetValue("balance");
var newBobBalance = bobBalance.AsInt32 + amountToTransfer;
accountColl.UpdateOne(session, Builders.Filter.Eq("name", "Bob"),
Builders.Update.Set("balance", newBobBalance));
bobBalance = accountColl.Find(session, Builders.Filter.Eq("name", "Bob")).FirstOrDefault().GetValue("balance");
Debug.Assert(bobBalance == newBobBalance);
}
catch (Exception e)
{
session.AbortTransaction();
throw;
}
session.CommitTransaction();
}
}
public void DoTransactionWithRetry(MongoClient client)
{
var dbName = "bank";
var collName = "account";
using (var session = client.StartSession(new ClientSessionOptions{CausalConsistency = false}))
{
try
{
var bankDB = client.GetDatabase(dbName);
var accountColl = bankDB.GetCollection(collName);
bankDB.DropCollection(collName);
accountColl.InsertOne(session, new BsonDocument { {"name", "Alice"}, {"balance", 1000 } });
accountColl.InsertOne(session, new BsonDocument { {"name", "Bob"}, {"balance", 1000 } });
while(true) {
try
{
TransferMoneyWithRetry(accountColl, session);
break;
}
catch (MongoException e)
{
if(e.HasErrorLabel("TransientTransactionError"))
{
continue;
}
else
{
throw;
}
}
}
// check values outside of transaction
var aliceNewBalance = accountColl.Find(Builders.Filter.Eq("name", "Alice")).FirstOrDefault().GetValue("balance");
var bobNewBalance = accountColl.Find(Builders.Filter.Eq("name", "Bob")).FirstOrDefault().GetValue("balance");
Debug.Assert(aliceNewBalance == 500);
Debug.Assert(bobNewBalance == 1500);
}
catch (Exception e)
{
Console.WriteLine("Error running transaction: " + e.Message);
}
}
}
```
------
#### [ Ruby ]
次のコードは、Ruby で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
# Ruby Core API
def transfer_money_w_retry(session, accountColl)
amountToTransfer = 500
session.start_transaction(read_concern: {level: :snapshot}, write_concern: {w: :majority})
# deduct $500 from Alice's account
aliceBalance = accountColl.find({"name"=>"Alice"}, :session=> session).first['balance']
assert aliceBalance >= amountToTransfer
newAliceBalance = aliceBalance - amountToTransfer
accountColl.update_one({"name"=>"Alice"}, { "$set" => {"balance"=>newAliceBalance} }, :session=> session)
aliceBalance = accountColl.find({"name"=>"Alice"}, :session=> session).first['balance']
assert_equal(newAliceBalance, aliceBalance)
# add $500 to Bob's account
bobBalance = accountColl.find({"name"=>"Bob"}, :session=> session).first['balance']
newBobBalance = bobBalance + amountToTransfer
accountColl.update_one({"name"=>"Bob"}, { "$set" => {"balance"=>newBobBalance} }, :session=> session)
bobBalance = accountColl.find({"name"=>"Bob"}, :session=> session).first['balance']
assert_equal(newBobBalance, bobBalance)
session.commit_transaction
end
def do_txn_w_retry(client)
dbName = "bank"
collName = "account"
session = client.start_session(:causal_consistency=> false)
bankDB = Mongo::Database.new(client, dbName)
accountColl = bankDB[collName]
accountColl.drop()
accountColl.insert_one({"name"=>"Alice", "balance"=>1000})
accountColl.insert_one({"name"=>"Bob", "balance"=>1000})
begin
transferMoneyWithRetry(session, accountColl)
puts "transaction committed"
rescue Mongo::Error => e
if e.label?('TransientTransactionError')
retry
else
puts "transaction failed"
raise
end
end
# check results outside of transaction
aliceBalance = accountColl.find({"name"=>"Alice"}).first['balance']
bobBalance = accountColl.find({"name"=>"Bob"}).first['balance']
assert_equal(aliceBalance, 500)
assert_equal(bobBalance, 1500)
end
```
------
#### [ Go ]
次のコードは、Go で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// Go - Core API
type Account struct {
Name string
Balance int
}
func transferMoneyWithRetry(sessionContext mongo.SessionContext, accountColl *mongo.Collection, t *testing.T) error {
amountToTransfer := 500
transactionOptions := options.Transaction().SetReadConcern(readconcern.Snapshot()).
SetWriteConcern(writeconcern.New(writeconcern.WMajority()))
if err := sessionContext.StartTransaction(transactionOptions); err != nil {
panic(err)
}
var result Account
// deduct $500 from Alice's account
err := accountColl.FindOne(sessionContext, bson.M{"name": "Alice"}).Decode(&result)
aliceBalance := result.Balance
newAliceBalance := aliceBalance - amountToTransfer
_, err = accountColl.UpdateOne(sessionContext, bson.M{"name": "Alice"}, bson.M{"$set": bson.M{"balance": newAliceBalance}})
if err != nil {
sessionContext.AbortTransaction(sessionContext)
}
err = accountColl.FindOne(sessionContext, bson.M{"name": "Alice"}).Decode(&result)
aliceBalance = result.Balance
assert.Equal(t, aliceBalance, newAliceBalance)
// add $500 to Bob's account
err = accountColl.FindOne(sessionContext, bson.M{"name": "Bob"}).Decode(&result)
bobBalance := result.Balance
newBobBalance := bobBalance + amountToTransfer
_, err = accountColl.UpdateOne(sessionContext, bson.M{"name": "Bob"}, bson.M{"$set": bson.M{"balance": newBobBalance}})
if err != nil {
sessionContext.AbortTransaction(sessionContext)
}
err = accountColl.FindOne(sessionContext, bson.M{"name": "Bob"}).Decode(&result)
bobBalance = result.Balance
assert.Equal(t, bobBalance, newBobBalance)
err = sessionContext.CommitTransaction(sessionContext)
return err
}
func doTransactionWithRetry(t *testing.T) {
ctx := context.TODO()
dbName := "bank"
collName := "account"
bankDB := client.Database(dbName)
accountColl := bankDB.Collection(collName)
client.UseSessionWithOptions(ctx, options.Session().SetCausalConsistency(false), func(sessionContext mongo.SessionContext) error {
accountColl.Drop(ctx)
accountColl.InsertOne(sessionContext, bson.M{"name" : "Alice", "balance":1000})
accountColl.InsertOne(sessionContext, bson.M{"name" : "Bob", "balance":1000})
for {
err := transferMoneyWithRetry(sessionContext, accountColl, t)
if err == nil {
println("transaction committed")
return nil
}
if mongoErr := err.(mongo.CommandError); mongoErr.HasErrorLabel("TransientTransactionError") {
continue
}
println("transaction failed")
return err
}
})
// check results outside of transaction
var result Account
accountColl.FindOne(ctx, bson.M{"name": "Alice"}).Decode(&result)
aliceBalance := result.Balance
assert.Equal(t, aliceBalance, 500)
accountColl.FindOne(ctx, bson.M{"name": "Bob"}).Decode(&result)
bobBalance := result.Balance
assert.Equal(t, bobBalance, 1500)
}
```
------
#### [ Java ]
次のコードは、Java で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// Java (sync) - Core API
public void transferMoneyWithRetry() {
// connect to server
MongoClientURI mongoURI = new MongoClientURI(uri);
MongoClient mongoClient = new MongoClient(mongoURI);
MongoDatabase bankDB = mongoClient.getDatabase("bank");
MongoCollection accountColl = bankDB.getCollection("account");
accountColl.drop();
// insert some sample data
accountColl.insertOne(new Document("name", "Alice").append("balance", 1000));
accountColl.insertOne(new Document("name", "Bob").append("balance", 1000));
while (true) {
try {
doTransferMoneyWithRetry(accountColl, mongoClient);
break;
} catch (MongoException e) {
if (e.hasErrorLabel(MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL)) {
continue;
} else {
throw e;
}
}
}
}
public void doTransferMoneyWithRetry(MongoCollection accountColl, MongoClient mongoClient) {
int amountToTransfer = 500;
TransactionOptions txnOptions = TransactionOptions.builder()
.readConcern(ReadConcern.SNAPSHOT)
.writeConcern(WriteConcern.MAJORITY)
.build();
ClientSessionOptions sessionOptions = ClientSessionOptions.builder().causallyConsistent(false).build();
try ( ClientSession clientSession = mongoClient.startSession(sessionOptions) ) {
clientSession.startTransaction(txnOptions);
// deduct $500 from Alice's account
List documentList = new ArrayList<>();
accountColl.find(clientSession, new Document("name", "Alice")).into(documentList);
int aliceBalance = (int) documentList.get(0).get("balance");
Assert.assertTrue(aliceBalance >= amountToTransfer);
int newAliceBalance = aliceBalance - amountToTransfer;
accountColl.updateOne(clientSession, new Document("name", "Alice"), new Document("$set", new Document("balance", newAliceBalance)));
// check Alice's new balance
documentList = new ArrayList<>();
accountColl.find(clientSession, new Document("name", "Alice")).into(documentList);
int updatedBalance = (int) documentList.get(0).get("balance");
Assert.assertEquals(updatedBalance, newAliceBalance);
// add $500 to Bob's account
documentList = new ArrayList<>();
accountColl.find(clientSession, new Document("name", "Bob")).into(documentList);
int bobBalance = (int) documentList.get(0).get("balance");
int newBobBalance = bobBalance + amountToTransfer;
accountColl.updateOne(clientSession, new Document("name", "Bob"), new Document("$set", new Document("balance", newBobBalance)));
// check Bob's new balance
documentList = new ArrayList<>();
accountColl.find(clientSession, new Document("name", "Bob")).into(documentList);
updatedBalance = (int) documentList.get(0).get("balance");
Assert.assertEquals(updatedBalance, newBobBalance);
// commit transaction
clientSession.commitTransaction();
}
}
// Java (async) -- Core API
public void transferMoneyWithRetry() {
// connect to the server
MongoClient mongoClient = MongoClients.create(uri);
MongoDatabase bankDB = mongoClient.getDatabase("bank");
MongoCollection accountColl = bankDB.getCollection("account");
SubscriberLatchWrapper dropCallback = new SubscriberLatchWrapper<>();
mongoClient.getDatabase("bank").drop().subscribe(dropCallback);
dropCallback.await();
// insert some sample data
SubscriberLatchWrapper insertionCallback = new SubscriberLatchWrapper<>();
accountColl.insertOne(new Document("name", "Alice").append("balance", 1000)).subscribe(insertionCallback);
insertionCallback.await();
insertionCallback = new SubscriberLatchWrapper<>();
accountColl.insertOne(new Document("name", "Bob").append("balance", 1000)).subscribe(insertionCallback);;
insertionCallback.await();
while (true) {
try {
doTransferMoneyWithRetry(accountColl, mongoClient);
break;
} catch (MongoException e) {
if (e.hasErrorLabel(MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL)) {
continue;
} else {
throw e;
}
}
}
}
public void doTransferMoneyWithRetry(MongoCollection accountColl, MongoClient mongoClient) {
int amountToTransfer = 500;
// start the transaction
TransactionOptions txnOptions = TransactionOptions.builder()
.readConcern(ReadConcern.SNAPSHOT)
.writeConcern(WriteConcern.MAJORITY)
.build();
ClientSessionOptions sessionOptions = ClientSessionOptions.builder().causallyConsistent(false).build();
SubscriberLatchWrapper sessionCallback = new SubscriberLatchWrapper<>();
mongoClient.startSession(sessionOptions).subscribe(sessionCallback);
ClientSession session = sessionCallback.get().get(0);
session.startTransaction(txnOptions);
// deduct $500 from Alice's account
SubscriberLatchWrapper findCallback = new SubscriberLatchWrapper<>();
accountColl.find(session, new Document("name", "Alice")).first().subscribe(findCallback);
Document documentFound = findCallback.get().get(0);
int aliceBalance = (int) documentFound.get("balance");
int newAliceBalance = aliceBalance - amountToTransfer;
SubscriberLatchWrapper updateCallback = new SubscriberLatchWrapper<>();
accountColl.updateOne(session, new Document("name", "Alice"), new Document("$set", new Document("balance", newAliceBalance))).subscribe(updateCallback);
updateCallback.await();
// check Alice's new balance
findCallback = new SubscriberLatchWrapper<>();
accountColl.find(session, new Document("name", "Alice")).first().subscribe(findCallback);
documentFound = findCallback.get().get(0);
int updatedBalance = (int) documentFound.get("balance");
Assert.assertEquals(updatedBalance, newAliceBalance);
// add $500 to Bob's account
findCallback = new SubscriberLatchWrapper<>();
accountColl.find(session, new Document("name", "Bob")).first().subscribe(findCallback);
documentFound = findCallback.get().get(0);
int bobBalance = (int) documentFound.get("balance");
int newBobBalance = bobBalance + amountToTransfer;
updateCallback = new SubscriberLatchWrapper<>();
accountColl.updateOne(session, new Document("name", "Bob"), new Document("$set", new Document("balance", newBobBalance))).subscribe(updateCallback);
updateCallback.await();
// check Bob's new balance
findCallback = new SubscriberLatchWrapper<>();
accountColl.find(session, new Document("name", "Bob")).first().subscribe(findCallback);
documentFound = findCallback.get().get(0);
updatedBalance = (int) documentFound.get("balance");
Assert.assertEquals(updatedBalance, newBobBalance);
// commit the transaction
SubscriberLatchWrapper transactionCallback = new SubscriberLatchWrapper<>();
session.commitTransaction().subscribe(transactionCallback);
transactionCallback.await();
}
public class SubscriberLatchWrapper implements Subscriber {
/**
* A Subscriber that stores the publishers results and provides a latch so can block on completion.
*
* @param The publishers result type
*/
private final List received;
private final List errors;
private final CountDownLatch latch;
private volatile Subscription subscription;
private volatile boolean completed;
/**
* Construct an instance
*/
public SubscriberLatchWrapper() {
this.received = new ArrayList<>();
this.errors = new ArrayList<>();
this.latch = new CountDownLatch(1);
}
@Override
public void onSubscribe(final Subscription s) {
subscription = s;
subscription.request(Integer.MAX_VALUE);
}
@Override
public void onNext(final T t) {
received.add(t);
}
@Override
public void onError(final Throwable t) {
if (t instanceof RuntimeException) {
errors.add((RuntimeException) t);
} else {
errors.add(new RuntimeException("Unexpected exception", t));
}
onComplete();
}
@Override
public void onComplete() {
completed = true;
subscription.cancel();
latch.countDown();
}
/**
* Get received elements
*
* @return the list of received elements
*/
public List getReceived() {
return received;
}
/**
* Get received elements.
*
* @return the list of receive elements
*/
public List get() {
return await().getReceived();
}
/**
* Await completion or error
*
* @return this
*/
public SubscriberLatchWrapper await() {
subscription.request(Integer.MAX_VALUE);
try {
if (!latch.await(300, TimeUnit.SECONDS)) {
throw new MongoTimeoutException("Publisher onComplete timed out for 300 seconds");
}
} catch (InterruptedException e) {
throw new MongoInterruptedException("Interrupted waiting for observeration", e);
}
if (!errors.isEmpty()) {
throw errors.get(0);
}
return this;
}
public boolean getCompleted() {
return this.completed;
}
public void close() {
subscription.cancel();
received.clear();
}
}
```
------
#### [ C ]
次のコードは、C で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// Sample C code with core session
bool core_session(mongoc_client_session_t *client_session, mongoc_collection_t* collection, bson_t *selector, int64_t balance){
bool r = true;
bson_error_t error;
bson_t *opts = bson_new();
bson_t *update = BCON_NEW ("$set", "{", "balance", BCON_INT64 (balance), "}");
// set read & write concern
mongoc_read_concern_t *read_concern = mongoc_read_concern_new ();
mongoc_write_concern_t *write_concern = mongoc_write_concern_new ();
mongoc_transaction_opt_t *txn_opts = mongoc_transaction_opts_new ();
mongoc_write_concern_set_w(write_concern, MONGOC_WRITE_CONCERN_W_MAJORITY);
mongoc_read_concern_set_level(read_concern, MONGOC_READ_CONCERN_LEVEL_SNAPSHOT);
mongoc_transaction_opts_set_write_concern (txn_opts, write_concern);
mongoc_transaction_opts_set_read_concern (txn_opts, read_concern);
mongoc_client_session_start_transaction (client_session, txn_opts, &error);
mongoc_client_session_append (client_session, opts, &error);
r = mongoc_collection_update_one (collection, selector, update, opts, NULL, &error);
mongoc_client_session_commit_transaction (client_session, NULL, &error);
bson_destroy (opts);
mongoc_transaction_opts_destroy(txn_opts);
mongoc_read_concern_destroy(read_concern);
mongoc_write_concern_destroy(write_concern);
bson_destroy (update);
return r;
}
void test_core_money_transfer(mongoc_client_t* client, mongoc_collection_t* collection, int amount_to_transfer){
bson_t reply;
bool r = true;
const bson_t *doc;
bson_iter_t iter;
bson_error_t error;
// find query
bson_t *alice_query = bson_new ();
BSON_APPEND_UTF8(alice_query, "name", "Alice");
bson_t *bob_query = bson_new ();
BSON_APPEND_UTF8(bob_query, "name", "Bob");
// create session
// set causal consistency to false
mongoc_session_opt_t *session_opts = mongoc_session_opts_new ();
mongoc_session_opts_set_causal_consistency (session_opts, false);
// start the session
mongoc_client_session_t *client_session = mongoc_client_start_session (client, session_opts, &error);
// add session to options
bson_t *opts = bson_new();
mongoc_client_session_append (client_session, opts, &error);
// deduct 500 from Alice
// find account balance of Alice
mongoc_cursor_t *cursor = mongoc_collection_find_with_opts (collection, alice_query, NULL, NULL);
mongoc_cursor_next (cursor, &doc);
bson_iter_init (&iter, doc);
bson_iter_find (&iter, "balance");
int64_t alice_balance = (bson_iter_value (&iter))->value.v_int64;
assert(alice_balance >= amount_to_transfer);
int64_t new_alice_balance = alice_balance - amount_to_transfer;
// core
r = core_session (client_session, collection, alice_query, new_alice_balance);
assert(r);
// find account balance of Alice after transaction
cursor = mongoc_collection_find_with_opts (collection, alice_query, NULL, NULL);
mongoc_cursor_next (cursor, &doc);
bson_iter_init (&iter, doc);
bson_iter_find (&iter, "balance");
alice_balance = (bson_iter_value (&iter))->value.v_int64;
assert(alice_balance == new_alice_balance);
assert(alice_balance == 500);
// add 500 to Bob's balance
// find account balance of Bob
cursor = mongoc_collection_find_with_opts (collection, bob_query, NULL, NULL);
mongoc_cursor_next (cursor, &doc);
bson_iter_init (&iter, doc);
bson_iter_find (&iter, "balance");
int64_t bob_balance = (bson_iter_value (&iter))->value.v_int64;
int64_t new_bob_balance = bob_balance + amount_to_transfer;
//core
r = core_session (client_session, collection, bob_query, new_bob_balance);
assert(r);
// find account balance of Bob after transaction
cursor = mongoc_collection_find_with_opts (collection, bob_query, NULL, NULL);
mongoc_cursor_next (cursor, &doc);
bson_iter_init (&iter, doc);
bson_iter_find (&iter, "balance");
bob_balance = (bson_iter_value (&iter))->value.v_int64;
assert(bob_balance == new_bob_balance);
assert(bob_balance == 1500);
// cleanup
bson_destroy(alice_query);
bson_destroy(bob_query);
mongoc_client_session_destroy(client_session);
bson_destroy(opts);
mongoc_cursor_destroy(cursor);
bson_destroy(doc);
}
int main(int argc, char* argv[]) {
mongoc_init ();
mongoc_client_t* client = mongoc_client_new ();
bson_error_t error;
// connect to bank db
mongoc_database_t *database = mongoc_client_get_database (client, "bank");
// access account collection
mongoc_collection_t* collection = mongoc_client_get_collection(client, "bank", "account");
// set amount to transfer
int64_t amount_to_transfer = 500;
// delete the collection if already existing
mongoc_collection_drop(collection, &error);
// open Alice account
bson_t *alice_account = bson_new ();
BSON_APPEND_UTF8(alice_account, "name", "Alice");
BSON_APPEND_INT64(alice_account, "balance", 1000);
// open Bob account
bson_t *bob_account = bson_new ();
BSON_APPEND_UTF8(bob_account, "name", "Bob");
BSON_APPEND_INT64(bob_account, "balance", 1000);
bool r = true;
r = mongoc_collection_insert_one(collection, alice_account, NULL, NULL, &error);
if (!r) {printf("Error encountered:%s", error.message);}
r = mongoc_collection_insert_one(collection, bob_account, NULL, NULL, &error);
if (!r) {printf("Error encountered:%s", error.message);}
test_core_money_transfer(client, collection, amount_to_transfer);
}
```
------
#### [ Scala ]
次のコードは、Scala で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// Scala Core API
def transferMoneyWithRetry(sessionObservable: SingleObservable[ClientSession] , database: MongoDatabase ): Unit = {
val accountColl = database.getCollection("account")
var amountToTransfer = 500
var transactionObservable: Observable[ClientSession] = sessionObservable.map(clientSession => {
clientSession.startTransaction()
// deduct $500 from Alice's account
var aliceBalance = accountColl.find(clientSession, Document("name" -> "Alice")).await().head.getInteger("balance")
assert(aliceBalance >= amountToTransfer)
var newAliceBalance = aliceBalance - amountToTransfer
accountColl.updateOne(clientSession, Document("name" -> "Alice"), Document("$set" -> Document("balance" -> newAliceBalance))).await()
aliceBalance = accountColl.find(clientSession, Document("name" -> "Alice")).await().head.getInteger("balance")
assert(aliceBalance == newAliceBalance)
// add $500 to Bob's account
var bobBalance = accountColl.find(clientSession, Document("name" -> "Bob")).await().head.getInteger("balance")
var newBobBalance = bobBalance + amountToTransfer
accountColl.updateOne(clientSession, Document("name" -> "Bob"), Document("$set" -> Document("balance" -> newBobBalance))).await()
bobBalance = accountColl.find(clientSession, Document("name" -> "Bob")).await().head.getInteger("balance")
assert(bobBalance == newBobBalance)
clientSession
})
transactionObservable.flatMap(clientSession => clientSession.commitTransaction()).await()
}
def doTransactionWithRetry(): Unit = {
val client: MongoClient = MongoClientWrapper.getMongoClient()
val database: MongoDatabase = client.getDatabase("bank")
val accountColl = database.getCollection("account")
accountColl.drop().await()
val sessionOptions = ClientSessionOptions.builder().causallyConsistent(false).build()
var sessionObservable: SingleObservable[ClientSession] = client.startSession(sessionOptions)
accountColl.insertOne(Document("name" -> "Alice", "balance" -> 1000)).await()
accountColl.insertOne(Document("name" -> "Bob", "balance" -> 1000)).await()
var retry = true
while (retry) {
try {
transferMoneyWithRetry(sessionObservable, database)
println("transaction committed")
retry = false
}
catch {
case e: MongoException if e.hasErrorLabel(MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL) => {
println("retrying transaction")
}
case other: Throwable => {
println("transaction failed")
retry = false
throw other
}
}
}
// check results outside of transaction
assert(accountColl.find(Document("name" -> "Alice")).results().head.getInteger("balance") == 500)
assert(accountColl.find(Document("name" -> "Bob")).results().head.getInteger("balance") == 1500)
accountColl.drop().await()
}
```
------
#### [ Python ]
次のコードは、Python で Amazon DocumentDB トランザクション API を使用する方法を示しています。
```
// Sample Python code with Core api
import pymongo
client = pymongo.MongoClient()
rc_snapshot = pymongo.read_concern.ReadConcern('snapshot')
wc_majority = pymongo.write_concern.WriteConcern('majority')
# To start, drop and create an account collection and insert balances for both Alice and Bob
collection = client.get_database("bank").get_collection("account")
collection.drop()
collection.insert_one({"_id": 1, "name": "Alice", "balance": 1000})
collection.insert_one({"_id": 2, "name": "Bob", "balance": 1000})
amount_to_transfer = 500
# deduct 500 from Alice's account
alice_balance = collection.find_one({"name": "Alice"}).get("balance")
assert alice_balance >= amount_to_transfer
new_alice_balance = alice_balance - amount_to_transfer
with client.start_session({'causalConsistency':False}) as session:
session.start_transaction(read_concern=rc_snapshot, write_concern=wc_majority)
collection.update_one({"name": "Alice"}, {'$set': {"balance": new_alice_balance}}, session=session)
session.commit_transaction()
updated_alice_balance = collection.find_one({"name": "Alice"}).get("balance")
assert updated_alice_balance == new_alice_balance
# add 500 to Bob's account
bob_balance = collection.find_one({"name": "Bob"}).get("balance")
assert bob_balance >= amount_to_transfer
new_bob_balance = bob_balance + amount_to_transfer
with client.start_session({'causalConsistency':False}) as session:
session.start_transaction(read_concern=rc_snapshot, write_concern=wc_majority)
collection.update_one({"name": "Bob"}, {'$set': {"balance": new_bob_balance}}, session=session)
session.commit_transaction()
updated_bob_balance = collection.find_one({"name": "Bob"}).get("balance")
assert updated_bob_balance == new_bob_balance
```
------
## サポートされているコマンド
| コマンド | サポート |
| --- | --- |
| `abortTransaction` | はい |
| `commitTransaction` | はい |
| `endSessions` | はい |
| `killSession` | はい |
| `killAllSession` | はい |
| `killAllSessionsByPattern` | いいえ |
| `refreshSessions` | いいえ |
| `startSession` | はい |
## サポート対象外の機能
| 方法 | ステージまたはコマンド |
| --- | --- |
| `db.collection.aggregate()` | `$collStats` `$currentOp` `$indexStats` `$listSessions` `$out` |
| `db.collection.count()` `db.collection.countDocuments()` | `$where` `$near` `$nearSphere` |
| `db.collection.insert()` | `insert` は、既存のコレクションに対して実行されない場合はサポートされません。このメソッドは、既存のコレクションを対象とする場合にサポートされます。 |
## セッション
MongoDB セッションは、再試行可能な書き込み、因果一貫性、トランザクション、およびデータベース間の操作の管理をサポートするために使用されるフレームワークです。セッションが作成されると、論理セッション識別子 (LSID) がクライアントによって生成され、サーバーにコマンドを送信するときに、そのセッション内のすべてのオペレーションにタグを付けるために使用されます。
Amazon DocumentDB は、トランザクションを有効にするためのセッションの使用をサポートしていますが、因果整合性や再試行可能な書き込みはサポートしていません。
Amazon DocumentDB 内でトランザクションを利用する場合、トランザクションはセッション内から `session.startTransaction()` API とセッションは、一度に 1 つのトランザクションをサポートします。同様に、トランザクションは commit (`session.commitTransaction()`) または中止 (`session.abortTransaction()`) API。
### 因果整合性
因果整合性により、単一のクライアントセッション内でクライアントが書き込み後の読み取り整合性、モノアトミックな読み取り/書き込みが読み取りに従うことが保証され、これらの保証はプライマリだけでなく、クラスタ内のすべてのインスタンスに適用されます。Amazon DocumentDB は因果一貫性をサポートしておらず、次のステートメントではエラーが発生します。
```
var mySession = db.getMongo().startSession();
var mySessionObject = mySession.getDatabase('test').getCollection('account');
mySessionObject.updateOne({"_id": 2}, {"$inc": {"balance": 400}});
//Result:{ "acknowledged" : true, "matchedCount" : 1, "modifiedCount" : 1 }
mySessionObject.find()
//Error: error: {
// "ok" : 0,
// "code" : 303,
// "errmsg" : "Feature not supported: 'causal consistency'",
// "operationTime" : Timestamp(1603461817, 493214)
//}
mySession.endSession()
```
セッション内で因果一貫性を無効にできます。そうすることで、セッションフレームワークを利用できますが、リードの因果一貫性の保証は提供されません。Amazon DocumentDB を使用する場合、プライマリからの読み取りは書き込み後の読み取り整合性になり、レプリカインスタンスからの読み取りは最終的に一貫します。トランザクションは、セッションを利用するための主要なユースケースです。
```
var mySession = db.getMongo().startSession({causalConsistency: false});
var mySessionObject = mySession.getDatabase('test').getCollection('account');
mySessionObject.updateOne({"_id": 2}, {"$inc": {"balance": 400}});
//Result:{ "acknowledged" : true, "matchedCount" : 1, "modifiedCount" : 1 }
mySessionObject.find()
//{ "_id" : 1, "name" : "Bob", "balance" : 100 }
//{ "_id" : 2, "name" : "Alice", "balance" : 1700 }
```
### 再試行可能な書き込み
再試行可能な書き込みは、ネットワークエラーが発生したとき、またはクライアントがプライマリを検出できない場合に、クライアントが書き込み操作を再試行する機能です。Amazon DocumentDB では、再試行可能な書き込みはサポートされていないため、無効にする必要があります。接続文字列のコマンド (`retryWrites=false`) で無効にできます。
**注記**
例外: mongo シェル を使用している場合は、どのコード文字列にも `retryWrites=false` コマンドを含めないでください。デフォルトでは、再試行可能な書き込みは無効になっています。`retryWrites=false` を含めると、通常の読み取りコマンドでは失敗する可能性があります。
## トランザクションエラー
トランザクションを使用する場合、トランザクション番号が進行中のトランザクションと一致しないというエラーを発生させるシナリオがあります。
このエラーは、少なくとも 2 つの異なるシナリオで生成できます。
+ 1 分間のトランザクションタイムアウト後。
+ インスタンスの再起動後 (パッチ適用、クラッシュリカバリなど)、トランザクションが正常にコミットされた場合でも、このエラーを受け取る可能性があります。インスタンスの再起動中、データベースは正常に完了したトランザクションと中止されたトランザクションの違いを判断できません。つまり、トランザクションの完了状態があいまいとなります。
このエラーを処理する最善の方法は、トランザクションの更新を冪等化することです。例えば、`$set` インクリメント/デクリメント操作の代わりにミューテーター。以下を参照してください。
```
{ "ok" : 0,
"operationTime" : Timestamp(1603938167, 1),
"code" : 251,
"errmsg" : "Given transaction number 1 does not match any in-progress transactions."
}
```