diff --git a/google/cloud/pubsub/internal/default_pull_ack_handler.h b/google/cloud/pubsub/internal/default_pull_ack_handler.h
index 54d414c45a87b..685494dac405f 100644
--- a/google/cloud/pubsub/internal/default_pull_ack_handler.h
+++ b/google/cloud/pubsub/internal/default_pull_ack_handler.h
@@ -23,11 +23,11 @@
 #include "google/cloud/future.h"
 #include "google/cloud/options.h"
 #include "google/cloud/status_or.h"
-#include "absl/types/optional.h"
 #include <chrono>
 #include <functional>
 #include <memory>
 #include <mutex>
+#include <optional>
 
 namespace google {
 namespace cloud {
diff --git a/google/cloud/pubsub/internal/defaults_test.cc b/google/cloud/pubsub/internal/defaults_test.cc
index 19e1ec41173c5..3a4fb8d94a67b 100644
--- a/google/cloud/pubsub/internal/defaults_test.cc
+++ b/google/cloud/pubsub/internal/defaults_test.cc
@@ -49,7 +49,7 @@ TEST(OptionsTest, SetEmulatorEnvOverrides) {
 }
 
 TEST(OptionsTest, UnsetEmulatorEnv) {
-  ScopedEnvironment emulator("PUBSUB_EMULATOR_HOST", absl::nullopt);
+  ScopedEnvironment emulator("PUBSUB_EMULATOR_HOST", std::nullopt);
   auto opts = DefaultCommonOptions(
       Options{}
           .set<EndpointOption>("used-endpoint")
@@ -129,14 +129,14 @@ TEST(OptionsTest, SetOtelLinkLimitEnvOverrides) {
 }
 
 TEST(OptionsTest, UnsetOtelLinkLimitEnv) {
-  ScopedEnvironment env("OTEL_SPAN_LINK_COUNT_LIMIT", absl::nullopt);
+  ScopedEnvironment env("OTEL_SPAN_LINK_COUNT_LIMIT", std::nullopt);
   auto opts =
       DefaultPublisherOptions(Options{}.set<pubsub::MaxOtelLinkCountOption>(1));
   EXPECT_EQ(1U, opts.get<pubsub::MaxOtelLinkCountOption>());
 }
 
 TEST(OptionsTest, UnsetOtelLinkLimitEnvNoUserOption) {
-  ScopedEnvironment env("OTEL_SPAN_LINK_COUNT_LIMIT", absl::nullopt);
+  ScopedEnvironment env("OTEL_SPAN_LINK_COUNT_LIMIT", std::nullopt);
   auto opts = DefaultPublisherOptions(Options{});
   EXPECT_EQ(128U, opts.get<pubsub::MaxOtelLinkCountOption>());
 }
diff --git a/google/cloud/pubsub/internal/streaming_subscription_batch_source.cc b/google/cloud/pubsub/internal/streaming_subscription_batch_source.cc
index 22009d1feaaec..4f30878bfd21f 100644
--- a/google/cloud/pubsub/internal/streaming_subscription_batch_source.cc
+++ b/google/cloud/pubsub/internal/streaming_subscription_batch_source.cc
@@ -325,10 +325,9 @@ void StreamingSubscriptionBatchSource::OnInitialWrite(RetryLoopState const& rs,
       shutdown_manager_->StartOperation(__func__, "InitialRead", [&] {
         auto weak = WeakFromThis();
         stream_->Read().then(
-            [weak,
-             rs](future<
-                 absl::optional<google::pubsub::v1::StreamingPullResponse>>
-                     f) {
+            [weak, rs](
+                future<std::optional<google::pubsub::v1::StreamingPullResponse>>
+                    f) {
               if (auto s = weak.lock())
                 s->OnInitialRead(std::move(rs), f.get());
             });
@@ -340,7 +339,7 @@ void StreamingSubscriptionBatchSource::OnInitialWrite(RetryLoopState const& rs,
 
 void StreamingSubscriptionBatchSource::OnInitialRead(
     RetryLoopState rs,
-    absl::optional<google::pubsub::v1::StreamingPullResponse> response) {
+    std::optional<google::pubsub::v1::StreamingPullResponse> response) {
   shutdown_manager_->FinishedOperation("InitialRead");
   if (!response.has_value()) {
     OnInitialError(std::move(rs));
@@ -426,15 +425,14 @@ void StreamingSubscriptionBatchSource::ReadLoop() {
   auto stream = stream_;
   lk.unlock();
   auto weak = WeakFromThis();
-  using ResponseType =
-      absl::optional<google::pubsub::v1::StreamingPullResponse>;
+  using ResponseType = std::optional<google::pubsub::v1::StreamingPullResponse>;
   stream->Read().then([weak, stream](future<ResponseType> f) {
     if (auto self = weak.lock()) self->OnRead(f.get());
   });
 }
 
 void StreamingSubscriptionBatchSource::OnRead(
-    absl::optional<google::pubsub::v1::StreamingPullResponse> response) {
+    std::optional<google::pubsub::v1::StreamingPullResponse> response) {
   auto weak = WeakFromThis();
   std::unique_lock<std::mutex> lk(mu_);
   pending_read_ = false;
diff --git a/google/cloud/pubsub/internal/streaming_subscription_batch_source.h b/google/cloud/pubsub/internal/streaming_subscription_batch_source.h
index e78b21c0d99ea..f90c65afe6cf5 100644
--- a/google/cloud/pubsub/internal/streaming_subscription_batch_source.h
+++ b/google/cloud/pubsub/internal/streaming_subscription_batch_source.h
@@ -103,7 +103,7 @@ class StreamingSubscriptionBatchSource
   void OnInitialWrite(RetryLoopState const& rs, bool ok);
   void OnInitialRead(
       RetryLoopState rs,
-      absl::optional<google::pubsub::v1::StreamingPullResponse> response);
+      std::optional<google::pubsub::v1::StreamingPullResponse> response);
   void OnInitialError(RetryLoopState rs);
   void OnInitialFinish(RetryLoopState rs, Status status);
   void OnBackoff(RetryLoopState rs, Status status);
@@ -112,7 +112,7 @@ class StreamingSubscriptionBatchSource
   void ReadLoop();
 
   void OnRead(
-      absl::optional<google::pubsub::v1::StreamingPullResponse> response);
+      std::optional<google::pubsub::v1::StreamingPullResponse> response);
   void ShutdownStream(std::unique_lock<std::mutex> lk, char const* reason);
   void OnFinish(Status status);
 
@@ -141,7 +141,7 @@ class StreamingSubscriptionBatchSource
   bool pending_read_ = false;
   Status status_;
   std::shared_ptr<AsyncPullStream> stream_;
-  absl::optional<bool> exactly_once_delivery_enabled_;
+  std::optional<bool> exactly_once_delivery_enabled_;
   std::vector<std::pair<std::string, std::chrono::seconds>> deadlines_queue_;
 };
 
diff --git a/google/cloud/pubsub/internal/streaming_subscription_batch_source_test.cc b/google/cloud/pubsub/internal/streaming_subscription_batch_source_test.cc
index 7ef2ddd51cec5..fadf8a1bfa11d 100644
--- a/google/cloud/pubsub/internal/streaming_subscription_batch_source_test.cc
+++ b/google/cloud/pubsub/internal/streaming_subscription_batch_source_test.cc
@@ -85,8 +85,8 @@ class FakeStream {
       return AddAction("Read").then([](future<bool> g) {
         auto ok = g.get();
         using Response = ::google::pubsub::v1::StreamingPullResponse;
-        if (!ok) return absl::optional<Response>{};
-        return absl::make_optional(Response{});
+        if (!ok) return std::optional<Response>{};
+        return std::make_optional(Response{});
       });
     };
     auto finish_response = [this] {
@@ -238,7 +238,7 @@ TEST(StreamingSubscriptionBatchSourceTest, StartTooManyTransientFailures) {
             };
         auto read_response = [cq]() mutable {
           return cq.MakeRelativeTimer(us(10)).then(
-              [](F) { return absl::optional<Response>{}; });
+              [](F) { return std::optional<Response>{}; });
         };
         auto finish_response = [cq, transient]() mutable {
           return cq.MakeRelativeTimer(us(10)).then(
@@ -303,7 +303,7 @@ TEST(StreamingSubscriptionBatchSourceTest, StartPermanentFailure) {
             };
         auto read_response = [cq]() mutable {
           return cq.MakeRelativeTimer(us(10)).then(
-              [](F) { return absl::optional<Response>{}; });
+              [](F) { return std::optional<Response>{}; });
         };
         auto finish_response = [cq, transient]() mutable {
           return cq.MakeRelativeTimer(us(10)).then(
@@ -459,12 +459,12 @@ TEST(StreamingSubscriptionBatchSourceTest, ResumeAfterFirstRead) {
             response.add_received_messages()->set_ack_id(
                 "ack-" + std::to_string(start + i));
           }
-          return absl::make_optional(std::move(response));
+          return std::make_optional(std::move(response));
         });
       };
       auto read_failure = [cq]() mutable {
         return cq.MakeRelativeTimer(us(10)).then(
-            [](F) { return absl::optional<Response>{}; });
+            [](F) { return std::optional<Response>{}; });
       };
       auto finish_response = [cq]() mutable {
         return cq.MakeRelativeTimer(us(10)).then([](F) mutable {
@@ -641,11 +641,11 @@ std::unique_ptr<pubsub_testing::MockAsyncPullStream> MakeExactlyOnceStream(
     return aseq.PushBack("Read").then([](future<bool> g) {
       auto ok = g.get();
       using Response = ::google::pubsub::v1::StreamingPullResponse;
-      if (!ok) return absl::optional<Response>{};
+      if (!ok) return std::optional<Response>{};
       Response response;
       response.mutable_subscription_properties()
           ->set_exactly_once_delivery_enabled(true);
-      return absl::make_optional(std::move(response));
+      return std::make_optional(std::move(response));
     });
   };
   auto finish_response = [&aseq, finish_status] {
@@ -832,8 +832,8 @@ TEST(StreamingSubscriptionBatchSourceTest, ShutdownWithPendingReadCancel) {
     };
     auto read_response = [&] {
       return async.PushBack("Read").then([](future<bool> f) {
-        if (f.get()) return absl::make_optional(Response{});
-        return absl::optional<Response>{};
+        if (f.get()) return std::make_optional(Response{});
+        return std::optional<Response>{};
       });
     };
     auto cancel = [&] { async.PushBack("Cancel"); };
@@ -912,21 +912,21 @@ TEST(StreamingSubscriptionBatchSourceTest, ExactlyOnceDeadlineStateChange) {
         auto read_response_with_eos = [&] {
           return aseq.PushBack("Read").then([](future<bool> g) {
             using Response = ::google::pubsub::v1::StreamingPullResponse;
-            if (!g.get()) return absl::optional<Response>{};
+            if (!g.get()) return std::optional<Response>{};
             Response response;
             response.mutable_subscription_properties()
                 ->set_exactly_once_delivery_enabled(true);
-            return absl::make_optional(std::move(response));
+            return std::make_optional(std::move(response));
           });
         };
         auto read_response_without_eos = [&] {
           return aseq.PushBack("Read").then([](future<bool> g) {
             using Response = ::google::pubsub::v1::StreamingPullResponse;
-            if (!g.get()) return absl::optional<Response>{};
+            if (!g.get()) return std::optional<Response>{};
             Response response;
             response.mutable_subscription_properties()
                 ->set_exactly_once_delivery_enabled(false);
-            return absl::make_optional(std::move(response));
+            return std::make_optional(std::move(response));
           });
         };
         auto finish_response = [&] {
@@ -1222,7 +1222,7 @@ std::unique_ptr<pubsub_testing::MockAsyncPullStream> MakeUnusedStream(
       response.mutable_subscription_properties()
           ->set_exactly_once_delivery_enabled(true);
     }
-    return make_ready_future(absl::make_optional(std::move(response)));
+    return make_ready_future(std::make_optional(std::move(response)));
   };
   auto finish_response = []() { return make_ready_future(Status{}); };
 
diff --git a/google/cloud/pubsub/internal/subscriber_connection_impl_test.cc b/google/cloud/pubsub/internal/subscriber_connection_impl_test.cc
index 3bc3149a611eb..b2b6c693e33c7 100644
--- a/google/cloud/pubsub/internal/subscriber_connection_impl_test.cc
+++ b/google/cloud/pubsub/internal/subscriber_connection_impl_test.cc
@@ -119,12 +119,12 @@ StreamingPullMock MakeAsyncStreamingPullMock(
     auto generator = std::make_shared<MessageGenerator>();
     auto read_response = [cq, generator]() mutable {
       return cq.MakeRelativeTimer(us(10)).then([generator](auto) {
-        return absl::make_optional(generator->Generate(10));
+        return std::make_optional(generator->Generate(10));
       });
     };
     auto canceled_response = [cq]() mutable {
       return cq.MakeRelativeTimer(us(10)).then(
-          [](auto) { return absl::optional<Response>{}; });
+          [](auto) { return std::optional<Response>{}; });
     };
     auto finish_response = [cq]() mutable {
       return cq.MakeRelativeTimer(us(10)).then([](auto) { return Status{}; });
diff --git a/google/cloud/pubsub/internal/subscription_session_test.cc b/google/cloud/pubsub/internal/subscription_session_test.cc
index f8ce1dde883c7..20907b1a77b79 100644
--- a/google/cloud/pubsub/internal/subscription_session_test.cc
+++ b/google/cloud/pubsub/internal/subscription_session_test.cc
@@ -139,14 +139,13 @@ void ScheduleCallbacks(int64_t ack_count, bool enable_open_telemetry) {
                                                 std::to_string(count));
             ++count;
           }
-          return cq.MakeRelativeTimer(us(10)).then([response](TimerFuture) {
-            return absl::make_optional(response);
-          });
+          return cq.MakeRelativeTimer(us(10)).then(
+              [response](TimerFuture) { return std::make_optional(response); });
         });
         EXPECT_CALL(*stream, Cancel).Times(1);
         EXPECT_CALL(*stream, Read).WillRepeatedly([&] {
           return cq.MakeRelativeTimer(us(10)).then([](TimerFuture) {
-            return absl::optional<google::pubsub::v1::StreamingPullResponse>{};
+            return std::optional<google::pubsub::v1::StreamingPullResponse>{};
           });
         });
         EXPECT_CALL(*stream, Finish).WillOnce([&] {
@@ -295,12 +294,12 @@ TEST(SubscriptionSessionTest, ScheduleCallbacksExactlyOnce) {
             ++count;
           }
           return cq.MakeRelativeTimer(us(10)).then(
-              [response](auto) { return absl::make_optional(response); });
+              [response](auto) { return std::make_optional(response); });
         });
         EXPECT_CALL(*stream, Cancel).Times(1);
         EXPECT_CALL(*stream, Read).WillRepeatedly([&] {
           return cq.MakeRelativeTimer(us(10)).then([](auto) {
-            return absl::optional<google::pubsub::v1::StreamingPullResponse>{};
+            return std::optional<google::pubsub::v1::StreamingPullResponse>{};
           });
         });
         EXPECT_CALL(*stream, Finish).WillOnce([&] {
@@ -872,8 +871,8 @@ TEST(SubscriptionSessionTest, FireAndForgetShutdown) {
     };
     auto read_response = [&] {
       return on_read.PushBack("Read").then([](future<bool> f) {
-        if (f.get()) return absl::make_optional(Response{});
-        return absl::optional<Response>{};
+        if (f.get()) return std::make_optional(Response{});
+        return std::optional<Response>{};
       });
     };
     auto finish_response = [&] {
diff --git a/google/cloud/pubsub/testing/fake_streaming_pull.cc b/google/cloud/pubsub/testing/fake_streaming_pull.cc
index b2a1dfd924fff..c517995eb2cf5 100644
--- a/google/cloud/pubsub/testing/fake_streaming_pull.cc
+++ b/google/cloud/pubsub/testing/fake_streaming_pull.cc
@@ -67,12 +67,12 @@ std::unique_ptr<pubsub_testing::MockAsyncPullStream> FakeAsyncStreamingPull(
   auto generator = std::make_shared<MessageGenerator>();
   auto read_response = [cq, generator]() mutable {
     return cq.MakeRelativeTimer(us(10)).then([generator](TimerFuture) {
-      return absl::make_optional(generator->Generate(10));
+      return std::make_optional(generator->Generate(10));
     });
   };
   auto canceled_response = [cq]() mutable {
     return cq.MakeRelativeTimer(us(10)).then(
-        [](TimerFuture) { return absl::optional<Response>{}; });
+        [](TimerFuture) { return std::optional<Response>{}; });
   };
   auto finish_response = [cq]() mutable {
     return cq.MakeRelativeTimer(us(10)).then(
diff --git a/google/cloud/pubsub/testing/mock_subscriber_stub.h b/google/cloud/pubsub/testing/mock_subscriber_stub.h
index f34ec7aacbd5a..0b92b269a29ed 100644
--- a/google/cloud/pubsub/testing/mock_subscriber_stub.h
+++ b/google/cloud/pubsub/testing/mock_subscriber_stub.h
@@ -143,7 +143,7 @@ class MockAsyncPullStream : public MockSubscriberStub::StreamingPullStream {
  public:
   MOCK_METHOD(void, Cancel, (), (override));
   MOCK_METHOD(future<bool>, Start, (), (override));
-  MOCK_METHOD(future<absl::optional<google::pubsub::v1::StreamingPullResponse>>,
+  MOCK_METHOD(future<std::optional<google::pubsub::v1::StreamingPullResponse>>,
               Read, (), (override));
   MOCK_METHOD(future<bool>, Write,
               (google::pubsub::v1::StreamingPullRequest const&,