diff --git a/google/cloud/storage/examples/storage_async_samples.cc b/google/cloud/storage/examples/storage_async_samples.cc
index da4f6c5f5fbf3..8bf4933c483b5 100644
--- a/google/cloud/storage/examples/storage_async_samples.cc
+++ b/google/cloud/storage/examples/storage_async_samples.cc
@@ -728,6 +728,42 @@ void CreateAndWriteAppendableObject(google::cloud::storage::AsyncClient& client,
   std::cout << "File successfully uploaded " << object.DebugString() << "\n";
 }
 
+void CreateAndWriteAppendableObjectWithChecksum(
+    google::cloud::storage::AsyncClient& client,
+    std::vector<std::string> const& argv) {
+  //! [create-and-write-appendable-object-with-checksum]
+  // [START storage_create_and_write_appendable_object_with_checksum]
+  namespace gcs = google::cloud::storage;
+  auto coro = [](gcs::AsyncClient& client, std::string bucket_name,
+                 std::string object_name)
+      -> google::cloud::future<google::storage::v2::Object> {
+    auto [writer, token] =
+        (co_await client.StartAppendableObjectUpload(
+             gcs::BucketName(std::move(bucket_name)), std::move(object_name)))
+            .value();
+    std::cout << "Appendable upload started for object " << object_name << "\n";
+
+    token = (co_await writer.Write(std::move(token),
+                                   gcs::WritePayload("Some data\n")))
+                .value();
+    std::cout << "Wrote some data.\n";
+
+    // Set the expected CRC32C checksum in the current options scope
+    // just before calling Finalize().
+    // Note: 548262564U is the pre-computed CRC32C checksum for the string "Some
+    // data\n". If the data changes, this checksum must be updated to match.
+    google::cloud::internal::OptionsSpan span(
+        google::cloud::Options{}.set<gcs::UseCrc32cValueOption>(548262564U));
+    co_return (co_await writer.Finalize(std::move(token))).value();
+  };
+  // [END storage_create_and_write_appendable_object_with_checksum]
+  //! [create-and-write-appendable-object-with-checksum]
+  // The example is easier to test and run if we call the coroutine and block
+  // until it completes..
+  auto const object = coro(client, argv.at(0), argv.at(1)).get();
+  std::cout << "File successfully uploaded " << object.DebugString() << "\n";
+}
+
 void PauseAndResumeAppendableUpload(google::cloud::storage::AsyncClient& client,
                                     std::vector<std::string> const& argv) {
   //! [pause-and-resume-appendable-upload]
@@ -1035,6 +1071,12 @@ void CreateAndWriteAppendableObject(google::cloud::storage::AsyncClient&,
                "coroutines\n";
 }
 
+void CreateAndWriteAppendableObjectWithChecksum(
+    google::cloud::storage::AsyncClient&, std::vector<std::string> const&) {
+  std::cerr << "AsyncClient::CreateAndWriteAppendableObjectWithChecksum() "
+               "example requires coroutines\n";
+}
+
 void PauseAndResumeAppendableUpload(google::cloud::storage::AsyncClient&,
                                     std::vector<std::string> const&) {
   std::cerr << "AsyncClient::PauseAndResumeAppendableUpload() example requires "
@@ -1505,6 +1547,8 @@ int main(int argc, char* argv[]) try {
 
       make_entry("create-and-write-appendable-object", {},
                  CreateAndWriteAppendableObject),
+      make_entry("create-and-write-appendable-object-with-checksum", {},
+                 CreateAndWriteAppendableObjectWithChecksum),
       make_entry("pause-and-resume-appendable-upload", {},
                  PauseAndResumeAppendableUpload),
       make_entry("finalize-appendable-object-upload", {},
diff --git a/google/cloud/storage/internal/async/writer_connection_impl.cc b/google/cloud/storage/internal/async/writer_connection_impl.cc
index 3acafa12d7dd8..88dd17fc9cf74 100644
--- a/google/cloud/storage/internal/async/writer_connection_impl.cc
+++ b/google/cloud/storage/internal/async/writer_connection_impl.cc
@@ -13,11 +13,14 @@
 // limitations under the License.
 
 #include "google/cloud/storage/internal/async/writer_connection_impl.h"
+#include "google/cloud/storage/async/options.h"
+#include "google/cloud/storage/hashing_options.h"
 #include "google/cloud/storage/internal/async/handle_redirect_error.h"
 #include "google/cloud/storage/internal/async/partial_upload.h"
 #include "google/cloud/storage/internal/async/write_payload_impl.h"
 #include "google/cloud/storage/internal/grpc/ctype_cord_workaround.h"
 #include "google/cloud/storage/internal/grpc/object_metadata_parser.h"
+#include "google/cloud/storage/internal/grpc/object_request_parser.h"
 #include "google/cloud/internal/make_status.h"
 
 namespace google {
@@ -138,10 +141,43 @@ AsyncWriterConnectionImpl::Finalize(storage::WritePayload payload) {
 
   auto p = WritePayloadImpl::GetImpl(payload);
   auto size = p.size();
-  auto action = request_.has_append_object_spec() ||
-                        request_.write_object_spec().appendable()
-                    ? PartialUpload::kFinalize
-                    : PartialUpload::kFinalizeWithChecksum;
+  auto is_append = request_.has_append_object_spec() ||
+                   request_.write_object_spec().appendable();
+  auto const& current_options = google::cloud::internal::CurrentOptions();
+  auto merged = google::cloud::internal::MergeOptions(
+      current_options, options_ ? *options_ : google::cloud::Options{});
+
+  // Default to letting the internal hash function compute and send the
+  // checksum.
+  auto action = PartialUpload::kFinalizeWithChecksum;
+
+  if (is_append) {
+    // For appendable uploads, the internal hash function only sees the chunks
+    // uploaded in this stream, not the full object. We use `kFinalize` to avoid
+    // sending this partial hash, which would otherwise fail validation.
+    if (merged.has<google::cloud::storage::UseCrc32cValueOption>()) {
+      write.mutable_object_checksums()->set_crc32c(
+          merged.get<google::cloud::storage::UseCrc32cValueOption>());
+    }
+    action = PartialUpload::kFinalize;
+  } else if (current_options
+                 .has<google::cloud::storage::UseCrc32cValueOption>() ||
+             current_options.has<google::cloud::storage::UseMD5ValueOption>()) {
+    // If the user specified a manual expected checksum dynamically at
+    // Finalize() via current_options, we manually inject it and use `kFinalize`
+    // so the internal hash function doesn't overwrite it with its own computed
+    // hash.
+    if (current_options.has<google::cloud::storage::UseCrc32cValueOption>()) {
+      write.mutable_object_checksums()->set_crc32c(
+          current_options.get<google::cloud::storage::UseCrc32cValueOption>());
+    }
+    if (current_options.has<google::cloud::storage::UseMD5ValueOption>()) {
+      write.mutable_object_checksums()->set_md5_hash(
+          current_options.get<google::cloud::storage::UseMD5ValueOption>());
+    }
+    action = PartialUpload::kFinalize;
+  }
+
   auto coro = PartialUpload::Call(impl_, hash_function_, std::move(write),
                                   std::move(p), std::move(action));
   return coro->Start().then([coro, size, this](auto f) mutable {
diff --git a/google/cloud/storage/internal/async/writer_connection_impl_test.cc b/google/cloud/storage/internal/async/writer_connection_impl_test.cc
index 03ecf7699cc2c..37a4f9cd9a30c 100644
--- a/google/cloud/storage/internal/async/writer_connection_impl_test.cc
+++ b/google/cloud/storage/internal/async/writer_connection_impl_test.cc
@@ -14,9 +14,11 @@
 
 #include "google/cloud/storage/internal/async/writer_connection_impl.h"
 #include "google/cloud/mocks/mock_async_streaming_read_write_rpc.h"
+#include "google/cloud/storage/async/options.h"
 #include "google/cloud/storage/internal/async/write_object.h"
 #include "google/cloud/storage/internal/crc32c.h"
 #include "google/cloud/storage/internal/grpc/ctype_cord_workaround.h"
+#include "google/cloud/storage/internal/grpc/object_metadata_parser.h"
 #include "google/cloud/storage/internal/hash_function_impl.h"
 #include "google/cloud/storage/options.h"
 #include "google/cloud/storage/testing/canonical_errors.h"
@@ -744,6 +746,238 @@ TEST(AsyncWriterConnectionTest, FinalizeAppendableNoChecksum) {
   next.first.set_value(true);
 }
 
+TEST(AsyncWriterConnectionTest, FinalizeAppendableWithExpectedChecksum) {
+  AsyncSequencer<bool> sequencer;
+  auto mock = std::make_unique<MockStream>();
+  EXPECT_CALL(*mock, Cancel).Times(1);
+  EXPECT_CALL(*mock, Write)
+      .WillOnce([&](Request const& request, grpc::WriteOptions wopt) {
+        EXPECT_TRUE(request.finish_write());
+        EXPECT_TRUE(wopt.is_last_message());
+        EXPECT_EQ(request.common_object_request_params().encryption_algorithm(),
+                  "test-only-algo");
+        EXPECT_TRUE(request.has_object_checksums());
+        EXPECT_EQ(request.object_checksums().crc32c(), 123456);
+        return sequencer.PushBack("Write");
+      });
+  EXPECT_CALL(*mock, Read).WillOnce([&]() {
+    return sequencer.PushBack("Read").then([](auto f) {
+      if (!f.get()) return absl::optional<Response>();
+      return absl::make_optional(MakeTestResponse());
+    });
+  });
+  EXPECT_CALL(*mock, Finish).WillOnce([&] {
+    return sequencer.PushBack("Finish").then([](auto f) {
+      if (f.get()) return Status{};
+      return PermanentError();
+    });
+  });
+  auto hash = std::make_shared<MockHashFunction>();
+  EXPECT_CALL(*hash, Update(_, An<absl::Cord const&>(), _)).Times(1);
+  EXPECT_CALL(*hash, Finish).Times(0);
+
+  auto request = MakeRequest();
+  request.mutable_write_object_spec()->set_appendable(true);
+
+  auto options = internal::MakeImmutableOptions(
+      Options{}.set<storage::UseCrc32cValueOption>(123456));
+
+  auto tested = std::make_unique<AsyncWriterConnectionImpl>(
+      options, std::move(request), std::move(mock), hash, 1024);
+  auto response = tested->Finalize(WritePayload{});
+
+  auto next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Write");
+  next.first.set_value(true);
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Read");
+  next.first.set_value(true);
+  auto object = response.get();
+  EXPECT_THAT(object, IsOkAndHolds(IsProtoEqual(MakeTestObject())))
+      << "=" << object->DebugString();
+
+  tested = {};
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Finish");
+  next.first.set_value(true);
+}
+
+TEST(AsyncWriterConnectionTest,
+     FinalizeAppendableWithExpectedChecksumFromCurrentOptions) {
+  AsyncSequencer<bool> sequencer;
+  auto mock = std::make_unique<MockStream>();
+  EXPECT_CALL(*mock, Cancel).Times(1);
+  EXPECT_CALL(*mock, Write)
+      .WillOnce([&](Request const& request, grpc::WriteOptions wopt) {
+        EXPECT_TRUE(request.finish_write());
+        EXPECT_TRUE(wopt.is_last_message());
+        EXPECT_EQ(request.common_object_request_params().encryption_algorithm(),
+                  "test-only-algo");
+        EXPECT_TRUE(request.has_object_checksums());
+        EXPECT_EQ(request.object_checksums().crc32c(), 654321);
+        return sequencer.PushBack("Write");
+      });
+  EXPECT_CALL(*mock, Read).WillOnce([&]() {
+    return sequencer.PushBack("Read").then([](auto f) {
+      if (!f.get()) return absl::optional<Response>();
+      return absl::make_optional(MakeTestResponse());
+    });
+  });
+  EXPECT_CALL(*mock, Finish).WillOnce([&] {
+    return sequencer.PushBack("Finish").then([](auto f) {
+      if (f.get()) return Status{};
+      return PermanentError();
+    });
+  });
+  auto hash = std::make_shared<MockHashFunction>();
+  EXPECT_CALL(*hash, Update(_, An<absl::Cord const&>(), _)).Times(1);
+  // It shouldn't call Finish() because we use kFinalize!
+  EXPECT_CALL(*hash, Finish).Times(0);
+
+  auto request = MakeRequest();
+  request.mutable_write_object_spec()->set_appendable(true);
+
+  auto tested = std::make_unique<AsyncWriterConnectionImpl>(
+      TestOptions(), std::move(request), std::move(mock), hash, 1024);
+
+  internal::OptionsSpan span(
+      Options{}.set<storage::UseCrc32cValueOption>(654321));
+  auto response = tested->Finalize(WritePayload{});
+
+  auto next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Write");
+  next.first.set_value(true);
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Read");
+  next.first.set_value(true);
+  auto object = response.get();
+  EXPECT_THAT(object, IsOkAndHolds(IsProtoEqual(MakeTestObject())))
+      << "=" << object->DebugString();
+
+  tested = {};
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Finish");
+  next.first.set_value(true);
+}
+
+TEST(AsyncWriterConnectionTest,
+     FinalizeNonAppendableWithExpectedChecksumFromCurrentOptions) {
+  AsyncSequencer<bool> sequencer;
+  auto mock = std::make_unique<MockStream>();
+  EXPECT_CALL(*mock, Cancel).Times(1);
+  EXPECT_CALL(*mock, Write)
+      .WillOnce([&](Request const& request, grpc::WriteOptions wopt) {
+        EXPECT_TRUE(request.finish_write());
+        EXPECT_TRUE(wopt.is_last_message());
+        EXPECT_EQ(request.common_object_request_params().encryption_algorithm(),
+                  "test-only-algo");
+        EXPECT_TRUE(request.has_object_checksums());
+        EXPECT_EQ(request.object_checksums().crc32c(), 654321);
+        return sequencer.PushBack("Write");
+      });
+  EXPECT_CALL(*mock, Read).WillOnce([&]() {
+    return sequencer.PushBack("Read").then([](auto f) {
+      if (!f.get()) return absl::optional<Response>();
+      return absl::make_optional(MakeTestResponse());
+    });
+  });
+  EXPECT_CALL(*mock, Finish).WillOnce([&] {
+    return sequencer.PushBack("Finish").then([](auto f) {
+      if (f.get()) return Status{};
+      return PermanentError();
+    });
+  });
+  auto hash = std::make_shared<MockHashFunction>();
+  EXPECT_CALL(*hash, Update(_, An<absl::Cord const&>(), _)).Times(1);
+  // It shouldn't call Finish() because we use kFinalize!
+  EXPECT_CALL(*hash, Finish).Times(0);
+
+  auto request = MakeRequest();
+  // Ensure it's explicitly not appendable.
+  request.mutable_write_object_spec()->set_appendable(false);
+
+  auto tested = std::make_unique<AsyncWriterConnectionImpl>(
+      TestOptions(), std::move(request), std::move(mock), hash, 1024);
+
+  internal::OptionsSpan span(
+      Options{}.set<storage::UseCrc32cValueOption>(654321));
+  auto response = tested->Finalize(WritePayload{});
+
+  auto next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Write");
+  next.first.set_value(true);
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Read");
+  next.first.set_value(true);
+  auto object = response.get();
+  EXPECT_THAT(object, IsOkAndHolds(IsProtoEqual(MakeTestObject())))
+      << "=" << object->DebugString();
+
+  tested = {};
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Finish");
+  next.first.set_value(true);
+}
+
+TEST(AsyncWriterConnectionTest,
+     FinalizeNonAppendableWithExpectedMD5FromCurrentOptions) {
+  AsyncSequencer<bool> sequencer;
+  auto mock = std::make_unique<MockStream>();
+  EXPECT_CALL(*mock, Cancel).Times(1);
+  EXPECT_CALL(*mock, Write)
+      .WillOnce([&](Request const& request, grpc::WriteOptions wopt) {
+        EXPECT_TRUE(request.finish_write());
+        EXPECT_TRUE(wopt.is_last_message());
+        EXPECT_EQ(request.common_object_request_params().encryption_algorithm(),
+                  "test-only-algo");
+        EXPECT_TRUE(request.has_object_checksums());
+        EXPECT_EQ(request.object_checksums().md5_hash(), "test-md5");
+        return sequencer.PushBack("Write");
+      });
+  EXPECT_CALL(*mock, Read).WillOnce([&]() {
+    return sequencer.PushBack("Read").then([](auto f) {
+      if (!f.get()) return absl::optional<Response>();
+      return absl::make_optional(MakeTestResponse());
+    });
+  });
+  EXPECT_CALL(*mock, Finish).WillOnce([&] {
+    return sequencer.PushBack("Finish").then([](auto f) {
+      if (f.get()) return Status{};
+      return PermanentError();
+    });
+  });
+  auto hash = std::make_shared<MockHashFunction>();
+  EXPECT_CALL(*hash, Update(_, An<absl::Cord const&>(), _)).Times(1);
+  // It shouldn't call Finish() because we use kFinalize!
+  EXPECT_CALL(*hash, Finish).Times(0);
+
+  auto request = MakeRequest();
+  // Ensure it's explicitly not appendable.
+  request.mutable_write_object_spec()->set_appendable(false);
+
+  auto tested = std::make_unique<AsyncWriterConnectionImpl>(
+      TestOptions(), std::move(request), std::move(mock), hash, 1024);
+
+  internal::OptionsSpan span(
+      Options{}.set<storage::UseMD5ValueOption>("test-md5"));
+  auto response = tested->Finalize(WritePayload{});
+
+  auto next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Write");
+  next.first.set_value(true);
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Read");
+  next.first.set_value(true);
+  auto object = response.get();
+  EXPECT_THAT(object, IsOkAndHolds(IsProtoEqual(MakeTestObject())))
+      << "=" << object->DebugString();
+
+  tested = {};
+  next = sequencer.PopFrontWithName();
+  ASSERT_THAT(next.second, "Finish");
+  next.first.set_value(true);
+}
+
 TEST(AsyncWriterConnectionTest, ResumeWithHandle) {
   AsyncSequencer<bool> sequencer;
   auto mock = std::make_unique<MockStream>();