@@ -221,6 +221,8 @@ public void testCommitAborted() {
221
221
AbortInterceptor interceptor = new AbortInterceptor (0 );
222
222
try (ITConnection connection =
223
223
createConnection (interceptor , new CountTransactionRetryListener ())) {
224
+ interceptor .setUsingMultiplexedSession (
225
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
224
226
ApiFuture <Long > count = getTestRecordCountAsync (connection );
225
227
// do an insert
226
228
ApiFuture <Long > updateCount =
@@ -253,6 +255,8 @@ public void testInsertAborted() {
253
255
AbortInterceptor interceptor = new AbortInterceptor (0 );
254
256
try (ITConnection connection =
255
257
createConnection (interceptor , new CountTransactionRetryListener ())) {
258
+ interceptor .setUsingMultiplexedSession (
259
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
256
260
ApiFuture <Long > count = getTestRecordCountAsync (connection );
257
261
// indicate that the next statement should abort
258
262
interceptor .setProbability (1.0 );
@@ -276,6 +280,8 @@ public void testUpdateAborted() {
276
280
AbortInterceptor interceptor = new AbortInterceptor (0 );
277
281
try (ITConnection connection =
278
282
createConnection (interceptor , new CountTransactionRetryListener ())) {
283
+ interceptor .setUsingMultiplexedSession (
284
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
279
285
ApiFuture <Long > count = getTestRecordCountAsync (connection );
280
286
// insert a test record
281
287
connection .executeUpdateAsync (
@@ -309,6 +315,8 @@ public void testQueryAborted() {
309
315
AbortInterceptor interceptor = new AbortInterceptor (0 );
310
316
try (ITConnection connection =
311
317
createConnection (interceptor , new CountTransactionRetryListener ())) {
318
+ interceptor .setUsingMultiplexedSession (
319
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
312
320
// insert a test record
313
321
connection .executeUpdateAsync (
314
322
Statement .of ("INSERT INTO TEST (ID, NAME) VALUES (1, 'test aborted')" ));
@@ -359,6 +367,8 @@ public void testNextCallAborted() {
359
367
AbortInterceptor interceptor = new AbortInterceptor (0 );
360
368
try (ITConnection connection =
361
369
createConnection (interceptor , new CountTransactionRetryListener ())) {
370
+ interceptor .setUsingMultiplexedSession (
371
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
362
372
// insert two test records
363
373
connection .executeUpdateAsync (
364
374
Statement .of ("INSERT INTO TEST (ID, NAME) VALUES (1, 'test 1')" ));
@@ -392,6 +402,8 @@ public void testMultipleAborts() {
392
402
AbortInterceptor interceptor = new AbortInterceptor (0 );
393
403
try (ITConnection connection =
394
404
createConnection (interceptor , new CountTransactionRetryListener ())) {
405
+ interceptor .setUsingMultiplexedSession (
406
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
395
407
ApiFuture <Long > count = getTestRecordCountAsync (connection );
396
408
// do three inserts which all will abort and retry
397
409
interceptor .setProbability (1.0 );
@@ -428,6 +440,8 @@ public void testAbortAfterSelect() {
428
440
AbortInterceptor interceptor = new AbortInterceptor (0 );
429
441
try (ITConnection connection =
430
442
createConnection (interceptor , new CountTransactionRetryListener ())) {
443
+ interceptor .setUsingMultiplexedSession (
444
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
431
445
ApiFuture <Long > count = getTestRecordCountAsync (connection );
432
446
// insert a test record
433
447
connection .executeUpdateAsync (
@@ -504,6 +518,8 @@ public void testAbortWithResultSetHalfway() {
504
518
AbortInterceptor interceptor = new AbortInterceptor (0 );
505
519
try (ITConnection connection =
506
520
createConnection (interceptor , new CountTransactionRetryListener ())) {
521
+ interceptor .setUsingMultiplexedSession (
522
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
507
523
// insert two test records
508
524
connection .executeUpdateAsync (
509
525
Statement .of ("INSERT INTO TEST (ID, NAME) VALUES (1, 'test 1')" ));
@@ -539,6 +555,8 @@ public void testAbortWithResultSetFullyConsumed() {
539
555
AbortInterceptor interceptor = new AbortInterceptor (0 );
540
556
try (ITConnection connection =
541
557
createConnection (interceptor , new CountTransactionRetryListener ())) {
558
+ interceptor .setUsingMultiplexedSession (
559
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
542
560
// insert two test records
543
561
connection .executeUpdateAsync (
544
562
Statement .of ("INSERT INTO TEST (ID, NAME) VALUES (1, 'test 1')" ));
@@ -581,6 +599,8 @@ public void testAbortWithConcurrentInsert() {
581
599
AbortInterceptor interceptor = new AbortInterceptor (0 );
582
600
try (ITConnection connection =
583
601
createConnection (interceptor , new CountTransactionRetryListener ())) {
602
+ interceptor .setUsingMultiplexedSession (
603
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
584
604
// insert two test records
585
605
connection .executeUpdateAsync (
586
606
Statement .of ("INSERT INTO TEST (ID, NAME) VALUES (1, 'test 1')" ));
@@ -632,6 +652,8 @@ public void testAbortWithConcurrentDelete() {
632
652
AbortInterceptor interceptor = new AbortInterceptor (0 );
633
653
// first insert two test records
634
654
try (ITConnection connection = createConnection ()) {
655
+ interceptor .setUsingMultiplexedSession (
656
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
635
657
connection .executeUpdateAsync (
636
658
Statement .of ("INSERT INTO TEST (ID, NAME) VALUES (1, 'test 1')" ));
637
659
connection .executeUpdateAsync (
@@ -641,6 +663,8 @@ public void testAbortWithConcurrentDelete() {
641
663
// open a new connection and select the two test records
642
664
try (ITConnection connection =
643
665
createConnection (interceptor , new CountTransactionRetryListener ())) {
666
+ interceptor .setUsingMultiplexedSession (
667
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
644
668
// select the test records and consume the entire result set
645
669
try (AsyncResultSet rs =
646
670
connection .executeQueryAsync (Statement .of ("SELECT * FROM TEST ORDER BY ID" ))) {
@@ -694,6 +718,8 @@ public void testAbortWithConcurrentUpdate() {
694
718
// open a new connection and select the two test records
695
719
try (ITConnection connection =
696
720
createConnection (interceptor , new CountTransactionRetryListener ())) {
721
+ interceptor .setUsingMultiplexedSession (
722
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
697
723
// select the test records and consume the entire result set
698
724
try (AsyncResultSet rs =
699
725
connection .executeQueryAsync (Statement .of ("SELECT * FROM TEST ORDER BY ID" ))) {
@@ -744,6 +770,8 @@ public void testAbortWithUnseenConcurrentInsert() throws InterruptedException {
744
770
AbortInterceptor interceptor = new AbortInterceptor (0 );
745
771
try (ITConnection connection =
746
772
createConnection (interceptor , new CountTransactionRetryListener ())) {
773
+ interceptor .setUsingMultiplexedSession (
774
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
747
775
// insert three test records
748
776
connection .executeUpdateAsync (
749
777
Statement .of ("INSERT INTO TEST (ID, NAME) VALUES (1, 'test 1')" ));
@@ -833,6 +861,8 @@ public void testRetryLargeResultSet() {
833
861
final long UPDATED_RECORDS = 1000L ;
834
862
AbortInterceptor interceptor = new AbortInterceptor (0 );
835
863
try (ITConnection connection = createConnection ()) {
864
+ interceptor .setUsingMultiplexedSession (
865
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
836
866
// insert test records
837
867
for (int i = 0 ; i < NUMBER_OF_TEST_RECORDS ; i ++) {
838
868
connection .bufferedWrite (
@@ -845,6 +875,8 @@ public void testRetryLargeResultSet() {
845
875
}
846
876
try (ITConnection connection =
847
877
createConnection (interceptor , new CountTransactionRetryListener ())) {
878
+ interceptor .setUsingMultiplexedSession (
879
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
848
880
// select the test records and iterate over them
849
881
try (AsyncResultSet rs =
850
882
connection .executeQueryAsync (Statement .of ("SELECT * FROM TEST ORDER BY ID" ))) {
@@ -867,6 +899,8 @@ public void testRetryLargeResultSet() {
867
899
// Wait until the entire result set has been consumed.
868
900
get (finished );
869
901
}
902
+ interceptor .setUsingMultiplexedSession (
903
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
870
904
// Do an update that will abort and retry.
871
905
interceptor .setProbability (1.0 );
872
906
interceptor .setOnlyInjectOnce (true );
@@ -898,6 +932,8 @@ public void testRetryHighAbortRate() {
898
932
AbortInterceptor interceptor = new AbortInterceptor (0.25D );
899
933
try (ITConnection connection =
900
934
createConnection (interceptor , new CountTransactionRetryListener ())) {
935
+ interceptor .setUsingMultiplexedSession (
936
+ isMultiplexedSessionsEnabledForRW (connection .getSpanner ()));
901
937
// insert test records
902
938
for (int i = 0 ; i < NUMBER_OF_TEST_RECORDS ; i ++) {
903
939
connection .bufferedWrite (
0 commit comments