diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClient.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClient.java
index e30c648c36..8b2dcc3178 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClient.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/DatabaseClient.java
@@ -24,6 +24,7 @@
 import com.google.cloud.spanner.Statement.StatementFactory;
 import com.google.spanner.v1.BatchWriteResponse;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 
 /**
  * Interface for all the APIs that are used to read/write data into a Cloud Spanner database. An
@@ -417,6 +418,7 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
    *   <li>{@link Options#isolationLevel(IsolationLevel)}: The isolation level for the transaction
+   *   <li>{@link Options#readLockMode(ReadLockMode)}: The read lock mode for the transaction
    * </ul>
    */
   TransactionRunner readWriteTransaction(TransactionOption... options);
@@ -458,6 +460,7 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
    *   <li>{@link Options#isolationLevel(IsolationLevel)}: The isolation level for the transaction
+   *   <li>{@link Options#readLockMode(ReadLockMode)}: The read lock mode for the transaction
    * </ul>
    */
   TransactionManager transactionManager(TransactionOption... options);
@@ -499,6 +502,7 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
    *   <li>{@link Options#isolationLevel(IsolationLevel)}: The isolation level for the transaction
+   *   <li>{@link Options#readLockMode(ReadLockMode)}: The read lock mode for the transaction
    * </ul>
    */
   AsyncRunner runAsync(TransactionOption... options);
@@ -554,6 +558,7 @@ ServerStream<BatchWriteResponse> batchWriteAtLeastOnce(
    *   <li>{@link Options#commitStats()}: Request that the server includes commit statistics in the
    *       {@link CommitResponse}.
    *   <li>{@link Options#isolationLevel(IsolationLevel)}: The isolation level for the transaction
+   *   <li>{@link Options#readLockMode(ReadLockMode)}: The read lock mode for the transaction
    * </ul>
    */
   AsyncTransactionManager transactionManagerAsync(TransactionOption... options);
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Options.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Options.java
index c95b8d4bc2..1e6ce34d67 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Options.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/Options.java
@@ -22,6 +22,7 @@
 import com.google.spanner.v1.ReadRequest.OrderBy;
 import com.google.spanner.v1.RequestOptions.Priority;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import java.io.Serializable;
 import java.time.Duration;
 import java.util.Objects;
@@ -155,9 +156,50 @@ public static TransactionOption commitStats() {
    * process in the commit phase (when any needed locks are acquired). The validation process
    * succeeds only if there are no conflicting committed transactions (that committed mutations to
    * the read data at a commit timestamp after the read timestamp).
+   *
+   * @deprecated Use {@link Options#readLockMode(ReadLockMode)} instead.
    */
+  @Deprecated
   public static TransactionOption optimisticLock() {
-    return OPTIMISTIC_LOCK_OPTION;
+    return Options.readLockMode(ReadLockMode.OPTIMISTIC);
+  }
+
+  /**
+   * Returns a {@link TransactionOption} to set the desired {@link ReadLockMode} for a read-write
+   * transaction.
+   *
+   * <p>This option controls the locking behavior for read operations and queries within a
+   * read-write transaction. It works in conjunction with the transaction's {@link IsolationLevel}.
+   *
+   * <ul>
+   *   <li>{@link ReadLockMode#PESSIMISTIC}: Read locks are acquired immediately on read. This mode
+   *       only applies to {@code SERIALIZABLE} isolation. This mode prevents concurrent
+   *       modifications by locking data throughout the transaction. This reduces commit-time aborts
+   *       due to conflicts but can increase how long transactions wait for locks and the overall
+   *       contention.
+   *   <li>{@link ReadLockMode#OPTIMISTIC}: Locks for reads within the transaction are not acquired
+   *       on read. Instead the locks are acquired on commit to validate that read/queried data has
+   *       not changed since the transaction started. If a conflict is detected, the transaction
+   *       will fail. This mode only applies to {@code SERIALIZABLE} isolation. This mode defers
+   *       locking until commit, which can reduce contention and improve throughput. However, be
+   *       aware that this increases the risk of transaction aborts if there's significant write
+   *       competition on the same data.
+   *   <li>{@link ReadLockMode#READ_LOCK_MODE_UNSPECIFIED}: This is the default if no mode is set.
+   *       The locking behavior depends on the isolation level:
+   *       <ul>
+   *         <li>For {@code REPEATABLE_READ} isolation: Locking semantics default to {@code
+   *             OPTIMISTIC}. However, validation checks at commit are only performed for queries
+   *             using {@code SELECT FOR UPDATE}, statements with {@code LOCK_SCANNED_RANGES} hints,
+   *             and DML statements. <br>
+   *             Note: It is an error to explicitly set {@code ReadLockMode} when the isolation
+   *             level is {@code REPEATABLE_READ}.
+   *         <li>For all other isolation levels: If the read lock mode is not set, it defaults to
+   *             {@code PESSIMISTIC} locking.
+   *       </ul>
+   * </ul>
+   */
+  public static TransactionOption readLockMode(ReadLockMode readLockMode) {
+    return new ReadLockModeOption(readLockMode);
   }
 
   /**
@@ -367,16 +409,6 @@ void appendToOptions(Options options) {
     }
   }
 
-  /** Option to request Optimistic Concurrency Control for read/write transactions. */
-  static final class OptimisticLockOption extends InternalOption implements TransactionOption {
-    @Override
-    void appendToOptions(Options options) {
-      options.withOptimisticLock = true;
-    }
-  }
-
-  static final OptimisticLockOption OPTIMISTIC_LOCK_OPTION = new OptimisticLockOption();
-
   /** Option to request the transaction to be excluded from change streams. */
   static final class ExcludeTxnFromChangeStreamsOption extends InternalOption
       implements UpdateTransactionOption {
@@ -516,6 +548,20 @@ void appendToOptions(Options options) {
     }
   }
 
+  /** Option to set read lock mode for read/write transactions. */
+  static final class ReadLockModeOption extends InternalOption implements TransactionOption {
+    private final ReadLockMode readLockMode;
+
+    public ReadLockModeOption(ReadLockMode readLockMode) {
+      this.readLockMode = readLockMode;
+    }
+
+    @Override
+    void appendToOptions(Options options) {
+      options.readLockMode = readLockMode;
+    }
+  }
+
   private boolean withCommitStats;
 
   private Duration maxCommitDelay;
@@ -530,7 +576,6 @@ void appendToOptions(Options options) {
   private String tag;
   private String etag;
   private Boolean validateOnly;
-  private Boolean withOptimisticLock;
   private Boolean withExcludeTxnFromChangeStreams;
   private Boolean dataBoostEnabled;
   private DirectedReadOptions directedReadOptions;
@@ -540,6 +585,7 @@ void appendToOptions(Options options) {
   private Boolean lastStatement;
   private IsolationLevel isolationLevel;
   private XGoogSpannerRequestId reqId;
+  private ReadLockMode readLockMode;
 
   // Construction is via factory methods below.
   private Options() {}
@@ -644,10 +690,6 @@ Boolean validateOnly() {
     return validateOnly;
   }
 
-  Boolean withOptimisticLock() {
-    return withOptimisticLock;
-  }
-
   Boolean withExcludeTxnFromChangeStreams() {
     return withExcludeTxnFromChangeStreams;
   }
@@ -704,6 +746,10 @@ IsolationLevel isolationLevel() {
     return isolationLevel;
   }
 
+  ReadLockMode readLockMode() {
+    return readLockMode;
+  }
+
   @Override
   public String toString() {
     StringBuilder b = new StringBuilder();
@@ -740,9 +786,6 @@ public String toString() {
     if (validateOnly != null) {
       b.append("validateOnly: ").append(validateOnly).append(' ');
     }
-    if (withOptimisticLock != null) {
-      b.append("withOptimisticLock: ").append(withOptimisticLock).append(' ');
-    }
     if (withExcludeTxnFromChangeStreams != null) {
       b.append("withExcludeTxnFromChangeStreams: ")
           .append(withExcludeTxnFromChangeStreams)
@@ -772,6 +815,9 @@ public String toString() {
     if (reqId != null) {
       b.append("requestId: ").append(reqId.toString());
     }
+    if (readLockMode != null) {
+      b.append("readLockMode: ").append(readLockMode).append(' ');
+    }
     return b.toString();
   }
 
@@ -807,7 +853,6 @@ public boolean equals(Object o) {
         && Objects.equals(tag(), that.tag())
         && Objects.equals(etag(), that.etag())
         && Objects.equals(validateOnly(), that.validateOnly())
-        && Objects.equals(withOptimisticLock(), that.withOptimisticLock())
         && Objects.equals(withExcludeTxnFromChangeStreams(), that.withExcludeTxnFromChangeStreams())
         && Objects.equals(dataBoostEnabled(), that.dataBoostEnabled())
         && Objects.equals(directedReadOptions(), that.directedReadOptions())
@@ -815,7 +860,8 @@ public boolean equals(Object o) {
         && Objects.equals(isLastStatement(), that.isLastStatement())
         && Objects.equals(lockHint(), that.lockHint())
         && Objects.equals(isolationLevel(), that.isolationLevel())
-        && Objects.equals(reqId(), that.reqId());
+        && Objects.equals(reqId(), that.reqId())
+        && Objects.equals(readLockMode(), that.readLockMode());
   }
 
   @Override
@@ -857,9 +903,6 @@ public int hashCode() {
     if (validateOnly != null) {
       result = 31 * result + validateOnly.hashCode();
     }
-    if (withOptimisticLock != null) {
-      result = 31 * result + withOptimisticLock.hashCode();
-    }
     if (withExcludeTxnFromChangeStreams != null) {
       result = 31 * result + withExcludeTxnFromChangeStreams.hashCode();
     }
@@ -887,6 +930,9 @@ public int hashCode() {
     if (reqId != null) {
       result = 31 * result + reqId.hashCode();
     }
+    if (readLockMode != null) {
+      result = 31 * result + readLockMode.hashCode();
+    }
     return result;
   }
 
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionImpl.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionImpl.java
index bec93512a1..bce8a8d375 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionImpl.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SessionImpl.java
@@ -76,9 +76,6 @@ static TransactionOptions createReadWriteTransactionOptions(
       transactionOptions.setExcludeTxnFromChangeStreams(true);
     }
     TransactionOptions.ReadWrite.Builder readWrite = TransactionOptions.ReadWrite.newBuilder();
-    if (options.withOptimisticLock() == Boolean.TRUE) {
-      readWrite.setReadLockMode(TransactionOptions.ReadWrite.ReadLockMode.OPTIMISTIC);
-    }
     if (previousTransactionId != null
         && previousTransactionId != com.google.protobuf.ByteString.EMPTY) {
       readWrite.setMultiplexedSessionPreviousTransactionId(previousTransactionId);
@@ -86,6 +83,9 @@ static TransactionOptions createReadWriteTransactionOptions(
     if (options.isolationLevel() != null) {
       transactionOptions.setIsolationLevel(options.isolationLevel());
     }
+    if (options.readLockMode() != null) {
+      readWrite.setReadLockMode(options.readLockMode());
+    }
     transactionOptions.setReadWrite(readWrite);
     return transactionOptions.build();
   }
@@ -283,6 +283,9 @@ public CommitResponse writeAtLeastOnceWithOptions(
     if (options.isolationLevel() != null) {
       transactionOptionsBuilder.setIsolationLevel(options.isolationLevel());
     }
+    if (options.readLockMode() != null) {
+      transactionOptionsBuilder.getReadWriteBuilder().setReadLockMode(options.readLockMode());
+    }
     requestBuilder.setSingleUseTransaction(
         defaultTransactionOptions().toBuilder().mergeFrom(transactionOptionsBuilder.build()));
 
diff --git a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerOptions.java b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerOptions.java
index d789dcaa78..ebef8a92c8 100644
--- a/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerOptions.java
+++ b/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerOptions.java
@@ -69,6 +69,7 @@
 import com.google.spanner.v1.SpannerGrpc;
 import com.google.spanner.v1.TransactionOptions;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import io.grpc.CallCredentials;
 import io.grpc.CompressorRegistry;
 import io.grpc.Context;
@@ -1700,6 +1701,7 @@ public Builder setEnableEndToEndTracing(boolean enableEndToEndTracing) {
      * <pre>{@code
      * DefaultReadWriteTransactionOptions options = DefaultReadWriteTransactionOptions.newBuilder()
      * .setIsolationLevel(IsolationLevel.SERIALIZABLE)
+     * .setReadLockMode(ReadLockMode.OPTIMISTIC)
      * .build();
      * }</pre>
      */
@@ -1724,6 +1726,12 @@ public DefaultReadWriteTransactionOptionsBuilder setIsolationLevel(
           return this;
         }
 
+        public DefaultReadWriteTransactionOptionsBuilder setReadLockMode(
+            ReadLockMode readLockMode) {
+          transactionOptionsBuilder.getReadWriteBuilder().setReadLockMode(readLockMode);
+          return this;
+        }
+
         public DefaultReadWriteTransactionOptions build() {
           return new DefaultReadWriteTransactionOptions(transactionOptionsBuilder.build());
         }
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseClientImplWithDefaultRWTransactionOptionsTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseClientImplWithDefaultRWTransactionOptionsTest.java
index 634356f222..6be28413fa 100644
--- a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseClientImplWithDefaultRWTransactionOptionsTest.java
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/DatabaseClientImplWithDefaultRWTransactionOptionsTest.java
@@ -37,6 +37,7 @@
 import com.google.spanner.v1.ExecuteSqlRequest;
 import com.google.spanner.v1.ReadRequest;
 import com.google.spanner.v1.TransactionOptions.IsolationLevel;
+import com.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode;
 import io.grpc.Server;
 import io.grpc.Status;
 import io.grpc.inprocess.InProcessServerBuilder;
@@ -60,6 +61,10 @@ public class DatabaseClientImplWithDefaultRWTransactionOptionsTest {
       Options.isolationLevel(IsolationLevel.SERIALIZABLE);
   private static final TransactionOption RR_ISOLATION_OPTION =
       Options.isolationLevel(IsolationLevel.REPEATABLE_READ);
+  private static final TransactionOption OPTIMISTIC_READ_LOCK_OPTION =
+      Options.readLockMode(ReadLockMode.OPTIMISTIC);
+  private static final TransactionOption PESSIMISTIC_READ_LOCK_OPTION =
+      Options.readLockMode(ReadLockMode.PESSIMISTIC);
   private static final DatabaseId DATABASE_ID =
       DatabaseId.of("[PROJECT]", "[INSTANCE]", "[DATABASE]");
   private static MockSpannerServiceImpl mockSpanner;
@@ -68,10 +73,14 @@ public class DatabaseClientImplWithDefaultRWTransactionOptionsTest {
   private static LocalChannelProvider channelProvider;
   private Spanner spanner;
   private Spanner spannerWithRR;
+  private Spanner spannerWithRRPessimistic;
   private Spanner spannerWithSerializable;
+  private Spanner spannerWithSerOptimistic;
   private DatabaseClient client;
   private DatabaseClient clientWithRepeatableReadOption;
+  private DatabaseClient clientWithRRPessimisticOption;
   private DatabaseClient clientWithSerializableOption;
+  private DatabaseClient clientWithSerOptimisticOption;
 
   @BeforeClass
   public static void startStaticServer() throws IOException {
@@ -109,56 +118,97 @@ public static void stopServer() throws InterruptedException {
   public void setUp() {
     mockSpanner.reset();
     mockSpanner.removeAllExecutionTimes();
-    SpannerOptions.Builder spannerOptionsBuilder =
-        SpannerOptions.newBuilder()
-            .setProjectId("[PROJECT]")
-            .setChannelProvider(channelProvider)
-            .setCredentials(NoCredentials.getInstance());
-    spanner = spannerOptionsBuilder.build().getService();
-    spannerWithRR =
-        spannerOptionsBuilder
-            .setDefaultTransactionOptions(
-                DefaultReadWriteTransactionOptions.newBuilder()
-                    .setIsolationLevel(IsolationLevel.REPEATABLE_READ)
-                    .build())
+    spanner = getSpannerOptionsBuilder().build().getService();
+    spannerWithRR = getSpannerOptionsBuilder(IsolationLevel.REPEATABLE_READ).build().getService();
+    spannerWithRRPessimistic =
+        getSpannerOptionsBuilder(IsolationLevel.REPEATABLE_READ, ReadLockMode.PESSIMISTIC)
             .build()
             .getService();
     spannerWithSerializable =
-        spannerOptionsBuilder
-            .setDefaultTransactionOptions(
-                DefaultReadWriteTransactionOptions.newBuilder()
-                    .setIsolationLevel(IsolationLevel.SERIALIZABLE)
-                    .build())
+        getSpannerOptionsBuilder(IsolationLevel.SERIALIZABLE).build().getService();
+    spannerWithSerOptimistic =
+        getSpannerOptionsBuilder(IsolationLevel.SERIALIZABLE, ReadLockMode.OPTIMISTIC)
             .build()
             .getService();
     client = spanner.getDatabaseClient(DATABASE_ID);
     clientWithRepeatableReadOption = spannerWithRR.getDatabaseClient(DATABASE_ID);
+    clientWithRRPessimisticOption = spannerWithRRPessimistic.getDatabaseClient(DATABASE_ID);
     clientWithSerializableOption = spannerWithSerializable.getDatabaseClient(DATABASE_ID);
+    clientWithSerOptimisticOption = spannerWithSerOptimistic.getDatabaseClient(DATABASE_ID);
+  }
+
+  private static SpannerOptions.Builder getSpannerOptionsBuilder() {
+    return getSpannerOptionsBuilder(
+        IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED, ReadLockMode.READ_LOCK_MODE_UNSPECIFIED);
+  }
+
+  private static SpannerOptions.Builder getSpannerOptionsBuilder(IsolationLevel isolationLevel) {
+    return getSpannerOptionsBuilder(isolationLevel, ReadLockMode.READ_LOCK_MODE_UNSPECIFIED);
+  }
+
+  private static SpannerOptions.Builder getSpannerOptionsBuilder(
+      IsolationLevel isolationLevel, ReadLockMode readLockMode) {
+    SpannerOptions.Builder spannerOptionsBuilder =
+        SpannerOptions.newBuilder()
+            .setProjectId("[PROJECT]")
+            .setChannelProvider(channelProvider)
+            .setCredentials(NoCredentials.getInstance());
+    return spannerOptionsBuilder.setDefaultTransactionOptions(
+        DefaultReadWriteTransactionOptions.newBuilder()
+            .setIsolationLevel(isolationLevel)
+            .setReadLockMode(readLockMode)
+            .build());
   }
 
   private void executeTest(
       Consumer<DatabaseClient> testAction, IsolationLevel expectedIsolationLevel) {
     testAction.accept(client);
-    validateIsolationLevel(expectedIsolationLevel);
+    validateIsolationLevel(expectedIsolationLevel, ReadLockMode.READ_LOCK_MODE_UNSPECIFIED);
+  }
+
+  private void executeTest(
+      Consumer<DatabaseClient> testAction,
+      IsolationLevel expectedIsolationLevel,
+      ReadLockMode readLockMode) {
+    testAction.accept(client);
+    validateIsolationLevel(expectedIsolationLevel, readLockMode);
   }
 
   private void executeTestWithRR(
       Consumer<DatabaseClient> testAction, IsolationLevel expectedIsolationLevel) {
     testAction.accept(clientWithRepeatableReadOption);
-    validateIsolationLevel(expectedIsolationLevel);
+    validateIsolationLevel(expectedIsolationLevel, ReadLockMode.READ_LOCK_MODE_UNSPECIFIED);
+  }
+
+  private void executeTestWithRRPessimistic(
+      Consumer<DatabaseClient> testAction,
+      IsolationLevel expectedIsolationLevel,
+      ReadLockMode expectedReadLockMode) {
+    testAction.accept(clientWithRRPessimisticOption);
+    validateIsolationLevel(expectedIsolationLevel, expectedReadLockMode);
   }
 
   private void executeTestWithSerializable(
       Consumer<DatabaseClient> testAction, IsolationLevel expectedIsolationLevel) {
     testAction.accept(clientWithSerializableOption);
-    validateIsolationLevel(expectedIsolationLevel);
+    validateIsolationLevel(expectedIsolationLevel, ReadLockMode.READ_LOCK_MODE_UNSPECIFIED);
+  }
+
+  private void executeTestWithSerializableOptimistic(
+      Consumer<DatabaseClient> testAction,
+      IsolationLevel expectedIsolationLevel,
+      ReadLockMode expectedReadLockMode) {
+    testAction.accept(clientWithSerOptimisticOption);
+    validateIsolationLevel(expectedIsolationLevel, expectedReadLockMode);
   }
 
   @After
   public void tearDown() {
     spanner.close();
     spannerWithRR.close();
+    spannerWithRRPessimistic.close();
     spannerWithSerializable.close();
+    spannerWithSerOptimistic.close();
   }
 
   @Test
@@ -181,6 +231,16 @@ public void testWriteWithOptions_WithRRSpannerOptions() {
         IsolationLevel.REPEATABLE_READ);
   }
 
+  @Test
+  public void testWriteWithOptions_WithRRPessimisticSpannerOptions() {
+    executeTestWithRRPessimistic(
+        c ->
+            MockSpannerTestActions.writeInsertMutationWithOptions(
+                c, Options.priority(RpcPriority.HIGH)),
+        IsolationLevel.REPEATABLE_READ,
+        ReadLockMode.PESSIMISTIC);
+  }
+
   @Test
   public void testWriteWithOptions_WithSerializableTxnOption() {
     executeTestWithRR(
@@ -189,6 +249,16 @@ public void testWriteWithOptions_WithSerializableTxnOption() {
         IsolationLevel.SERIALIZABLE);
   }
 
+  @Test
+  public void testWriteWithOptions_WithSerializableOptimisticTxnOption() {
+    executeTestWithRRPessimistic(
+        c ->
+            MockSpannerTestActions.writeInsertMutationWithOptions(
+                c, SERIALIZABLE_ISOLATION_OPTION, OPTIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.SERIALIZABLE,
+        ReadLockMode.OPTIMISTIC);
+  }
+
   @Test
   public void testWriteAtLeastOnce_WithSerializableSpannerOptions() {
     executeTestWithSerializable(
@@ -204,6 +274,26 @@ public void testWriteAtLeastOnceWithOptions_WithRRTxnOption() {
         IsolationLevel.REPEATABLE_READ);
   }
 
+  @Test
+  public void testWriteAtLeastOnceWithOptions_WithRRPessimisticTxnOption() {
+    executeTestWithSerializableOptimistic(
+        c ->
+            MockSpannerTestActions.writeAtLeastOnceWithOptionsInsertMutation(
+                c, RR_ISOLATION_OPTION, PESSIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.REPEATABLE_READ,
+        ReadLockMode.PESSIMISTIC);
+  }
+
+  @Test
+  public void testWriteAtLeastOnceWithOptions_WithPessimisticTxnOption() {
+    executeTestWithRRPessimistic(
+        c ->
+            MockSpannerTestActions.writeAtLeastOnceWithOptionsInsertMutation(
+                c, OPTIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.REPEATABLE_READ,
+        ReadLockMode.OPTIMISTIC);
+  }
+
   @Test
   public void testReadWriteTxn_WithRRSpannerOption_batchUpdate() {
     executeTestWithRR(
@@ -217,6 +307,16 @@ public void testReadWriteTxn_WithSerializableTxnOption_batchUpdate() {
         IsolationLevel.SERIALIZABLE);
   }
 
+  @Test
+  public void testReadWriteTxn_WithSerOptimisticTxnOption_batchUpdate() {
+    executeTestWithRRPessimistic(
+        c ->
+            MockSpannerTestActions.executeBatchUpdateTransaction(
+                c, SERIALIZABLE_ISOLATION_OPTION, OPTIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.SERIALIZABLE,
+        ReadLockMode.OPTIMISTIC);
+  }
+
   @Test
   public void testPartitionedDML_WithRRSpannerOption() {
     executeTestWithRR(
@@ -231,6 +331,26 @@ public void testCommit_WithSerializableTxnOption() {
         IsolationLevel.SERIALIZABLE);
   }
 
+  @Test
+  public void testCommit_WithSerializablePessimisticTxnOption() {
+    executeTest(
+        c ->
+            MockSpannerTestActions.commitDeleteTransaction(
+                c, SERIALIZABLE_ISOLATION_OPTION, PESSIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.SERIALIZABLE,
+        ReadLockMode.PESSIMISTIC);
+  }
+
+  @Test
+  public void testCommit_WithSerializableOptimisticTxnOption() {
+    executeTest(
+        c ->
+            MockSpannerTestActions.commitDeleteTransaction(
+                c, SERIALIZABLE_ISOLATION_OPTION, OPTIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.SERIALIZABLE,
+        ReadLockMode.OPTIMISTIC);
+  }
+
   @Test
   public void testTransactionManagerCommit_WithRRTxnOption() {
     executeTestWithSerializable(
@@ -238,12 +358,28 @@ public void testTransactionManagerCommit_WithRRTxnOption() {
         IsolationLevel.REPEATABLE_READ);
   }
 
+  @Test
+  public void testTransactionManagerCommit_WithRRTxnOptionAndSerOptimisticSpannerOptions() {
+    executeTestWithSerializableOptimistic(
+        c -> MockSpannerTestActions.transactionManagerCommit(c, RR_ISOLATION_OPTION),
+        IsolationLevel.REPEATABLE_READ,
+        ReadLockMode.OPTIMISTIC);
+  }
+
   @Test
   public void testAsyncRunnerCommit_WithRRSpannerOption() {
     executeTestWithRR(
         c -> MockSpannerTestActions.asyncRunnerCommit(c, executor), IsolationLevel.REPEATABLE_READ);
   }
 
+  @Test
+  public void testAsyncRunnerCommit_WithSerOptimisticSpannerOption() {
+    executeTestWithSerializableOptimistic(
+        c -> MockSpannerTestActions.asyncRunnerCommit(c, executor),
+        IsolationLevel.SERIALIZABLE,
+        ReadLockMode.OPTIMISTIC);
+  }
+
   @Test
   public void testAsyncTransactionManagerCommit_WithSerializableTxnOption() {
     executeTestWithRR(
@@ -253,6 +389,24 @@ public void testAsyncTransactionManagerCommit_WithSerializableTxnOption() {
         IsolationLevel.SERIALIZABLE);
   }
 
+  @Test
+  public void testAsyncTransactionManagerCommit_WithRRPessimisticSpannerOptions() {
+    executeTestWithRRPessimistic(
+        c -> MockSpannerTestActions.transactionManagerAsyncCommit(c, executor),
+        IsolationLevel.REPEATABLE_READ,
+        ReadLockMode.PESSIMISTIC);
+  }
+
+  @Test
+  public void testAsyncTransactionManagerCommit_WithSerOptimisticTxnOption() {
+    executeTestWithRRPessimistic(
+        c ->
+            MockSpannerTestActions.transactionManagerAsyncCommit(
+                c, executor, SERIALIZABLE_ISOLATION_OPTION, OPTIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.SERIALIZABLE,
+        ReadLockMode.OPTIMISTIC);
+  }
+
   @Test
   public void testReadWriteTxn_WithNoOptions() {
     executeTest(MockSpannerTestActions::executeSelect1, IsolationLevel.ISOLATION_LEVEL_UNSPECIFIED);
@@ -265,6 +419,26 @@ public void executeSqlWithRWTransactionOptions_RepeatableRead() {
         IsolationLevel.REPEATABLE_READ);
   }
 
+  @Test
+  public void executeSqlWithRWTransactionOptions_RRPessimistic() {
+    executeTest(
+        c ->
+            MockSpannerTestActions.executeSelect1(
+                c, RR_ISOLATION_OPTION, PESSIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.REPEATABLE_READ,
+        ReadLockMode.PESSIMISTIC);
+  }
+
+  @Test
+  public void executeSqlWithRWTransactionOptions_RROptimistic() {
+    executeTest(
+        c ->
+            MockSpannerTestActions.executeSelect1(
+                c, RR_ISOLATION_OPTION, PESSIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.REPEATABLE_READ,
+        ReadLockMode.PESSIMISTIC);
+  }
+
   @Test
   public void
       executeSqlWithDefaultSpannerOptions_SerializableAndRWTransactionOptions_RepeatableRead() {
@@ -293,6 +467,26 @@ public void executeSqlWithRWTransactionOptions_Serializable() {
         IsolationLevel.SERIALIZABLE);
   }
 
+  @Test
+  public void executeSqlWithRWTransactionOptions_SerializablePessimistic() {
+    executeTest(
+        c ->
+            MockSpannerTestActions.executeSelect1(
+                c, SERIALIZABLE_ISOLATION_OPTION, PESSIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.SERIALIZABLE,
+        ReadLockMode.PESSIMISTIC);
+  }
+
+  @Test
+  public void executeSqlWithRWTransactionOptions_SerializableOptimistic() {
+    executeTest(
+        c ->
+            MockSpannerTestActions.executeSelect1(
+                c, SERIALIZABLE_ISOLATION_OPTION, OPTIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.SERIALIZABLE,
+        ReadLockMode.OPTIMISTIC);
+  }
+
   @Test
   public void readWithRWTransactionOptions_RepeatableRead() {
     executeTest(
@@ -300,6 +494,26 @@ public void readWithRWTransactionOptions_RepeatableRead() {
         IsolationLevel.REPEATABLE_READ);
   }
 
+  @Test
+  public void readWithRWTransactionOptions_RepeatableReadPessimistic() {
+    executeTest(
+        c ->
+            MockSpannerTestActions.executeReadFoo(
+                c, RR_ISOLATION_OPTION, PESSIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.REPEATABLE_READ,
+        ReadLockMode.PESSIMISTIC);
+  }
+
+  @Test
+  public void readWithRWTransactionOptions_RepeatableReadOptimistic() {
+    executeTest(
+        c ->
+            MockSpannerTestActions.executeReadFoo(
+                c, RR_ISOLATION_OPTION, OPTIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.REPEATABLE_READ,
+        ReadLockMode.OPTIMISTIC);
+  }
+
   @Test
   public void readWithRWTransactionOptions_Serializable() {
     executeTest(
@@ -321,33 +535,82 @@ public void beginTransactionWithRWTransactionOptions_Serializable() {
         IsolationLevel.SERIALIZABLE);
   }
 
-  private void validateIsolationLevel(IsolationLevel isolationLevel) {
+  @Test
+  public void beginTransactionWithRWTransactionOptions_RROptimistic() {
+    executeTestWithRRPessimistic(
+        c -> MockSpannerTestActions.executeInvalidAndValidSql(c, OPTIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.REPEATABLE_READ,
+        ReadLockMode.OPTIMISTIC);
+  }
+
+  @Test
+  public void beginTransactionWithRWTransactionOptions_SerPessimistic() {
+    executeTestWithRRPessimistic(
+        c -> MockSpannerTestActions.executeInvalidAndValidSql(c, SERIALIZABLE_ISOLATION_OPTION),
+        IsolationLevel.SERIALIZABLE,
+        ReadLockMode.PESSIMISTIC);
+  }
+
+  @Test
+  public void beginTransactionWithRWTransactionOptions_SerOptimistic() {
+    executeTestWithRRPessimistic(
+        c ->
+            MockSpannerTestActions.executeInvalidAndValidSql(
+                c, SERIALIZABLE_ISOLATION_OPTION, OPTIMISTIC_READ_LOCK_OPTION),
+        IsolationLevel.SERIALIZABLE,
+        ReadLockMode.OPTIMISTIC);
+  }
+
+  private void validateIsolationLevel(IsolationLevel isolationLevel, ReadLockMode readLockMode) {
     boolean foundMatchingRequest = false;
     for (AbstractMessage request : mockSpanner.getRequests()) {
       if (request instanceof ExecuteSqlRequest) {
         foundMatchingRequest = true;
         assertEquals(
-            ((ExecuteSqlRequest) request).getTransaction().getBegin().getIsolationLevel(),
-            isolationLevel);
+            isolationLevel,
+            ((ExecuteSqlRequest) request).getTransaction().getBegin().getIsolationLevel());
+        assertEquals(
+            readLockMode,
+            ((ExecuteSqlRequest) request)
+                .getTransaction()
+                .getBegin()
+                .getReadWrite()
+                .getReadLockMode());
       } else if (request instanceof BeginTransactionRequest) {
         foundMatchingRequest = true;
         assertEquals(
-            ((BeginTransactionRequest) request).getOptions().getIsolationLevel(), isolationLevel);
+            isolationLevel, ((BeginTransactionRequest) request).getOptions().getIsolationLevel());
+        assertEquals(
+            readLockMode,
+            ((BeginTransactionRequest) request).getOptions().getReadWrite().getReadLockMode());
       } else if (request instanceof ReadRequest) {
         foundMatchingRequest = true;
         assertEquals(
-            ((ReadRequest) request).getTransaction().getBegin().getIsolationLevel(),
-            isolationLevel);
+            isolationLevel,
+            ((ReadRequest) request).getTransaction().getBegin().getIsolationLevel());
+        assertEquals(
+            readLockMode,
+            ((ReadRequest) request).getTransaction().getBegin().getReadWrite().getReadLockMode());
       } else if (request instanceof CommitRequest) {
         foundMatchingRequest = true;
         assertEquals(
-            ((CommitRequest) request).getSingleUseTransaction().getIsolationLevel(),
-            isolationLevel);
+            isolationLevel,
+            ((CommitRequest) request).getSingleUseTransaction().getIsolationLevel());
+        assertEquals(
+            readLockMode,
+            ((CommitRequest) request).getSingleUseTransaction().getReadWrite().getReadLockMode());
       } else if (request instanceof ExecuteBatchDmlRequest) {
         foundMatchingRequest = true;
         assertEquals(
-            ((ExecuteBatchDmlRequest) request).getTransaction().getBegin().getIsolationLevel(),
-            isolationLevel);
+            isolationLevel,
+            ((ExecuteBatchDmlRequest) request).getTransaction().getBegin().getIsolationLevel());
+        assertEquals(
+            readLockMode,
+            ((ExecuteBatchDmlRequest) request)
+                .getTransaction()
+                .getBegin()
+                .getReadWrite()
+                .getReadLockMode());
       }
       if (foundMatchingRequest) {
         break;
diff --git a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/OptionsTest.java b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/OptionsTest.java
index 67a2adf123..8571c42b3d 100644
--- a/google-cloud-spanner/src/test/java/com/google/cloud/spanner/OptionsTest.java
+++ b/google-cloud-spanner/src/test/java/com/google/cloud/spanner/OptionsTest.java
@@ -735,6 +735,19 @@ public void optimisticLockEquality() {
     assertNotEquals(option1, option3);
   }
 
+  @Test
+  public void readLockModeEquality() {
+    Options option1 = Options.fromTransactionOptions(Options.readLockMode(ReadLockMode.OPTIMISTIC));
+    Options option2 = Options.fromTransactionOptions(Options.readLockMode(ReadLockMode.OPTIMISTIC));
+    Options option3 =
+        Options.fromTransactionOptions(Options.readLockMode(ReadLockMode.PESSIMISTIC));
+    Options option4 = Options.fromReadOptions();
+
+    assertEquals(option1, option2);
+    assertNotEquals(option1, option3);
+    assertNotEquals(option1, option4);
+  }
+
   @Test
   public void optimisticLockHashCode() {
     Options option1 = Options.fromTransactionOptions(Options.optimisticLock());
@@ -745,6 +758,19 @@ public void optimisticLockHashCode() {
     assertNotEquals(option1.hashCode(), option3.hashCode());
   }
 
+  @Test
+  public void readLockModeHashCode() {
+    Options option1 = Options.fromTransactionOptions(Options.readLockMode(ReadLockMode.OPTIMISTIC));
+    Options option2 = Options.fromTransactionOptions(Options.readLockMode(ReadLockMode.OPTIMISTIC));
+    Options option3 =
+        Options.fromTransactionOptions(Options.readLockMode(ReadLockMode.PESSIMISTIC));
+    Options option4 = Options.fromReadOptions();
+
+    assertEquals(option1.hashCode(), option2.hashCode());
+    assertNotEquals(option1.hashCode(), option3.hashCode());
+    assertNotEquals(option1.hashCode(), option4.hashCode());
+  }
+
   @Test
   public void directedReadEquality() {
     Options option1 = Options.fromReadOptions(Options.directedRead(DIRECTED_READ_OPTIONS));