diff --git a/quic/core/quic_datagram_queue.cc b/quic/core/quic_datagram_queue.cc
index e10a919..08c349c 100644
--- a/quic/core/quic_datagram_queue.cc
+++ b/quic/core/quic_datagram_queue.cc
@@ -17,7 +17,13 @@
 constexpr float kMinPacingWindows = 4;
 
 QuicDatagramQueue::QuicDatagramQueue(QuicSession* session)
-    : session_(session), clock_(session->connection()->clock()) {}
+    : QuicDatagramQueue(session, nullptr) {}
+
+QuicDatagramQueue::QuicDatagramQueue(QuicSession* session,
+                                     std::unique_ptr<Observer> observer)
+    : session_(session),
+      clock_(session->connection()->clock()),
+      observer_(std::move(observer)) {}
 
 MessageStatus QuicDatagramQueue::SendOrQueueDatagram(QuicMemSlice datagram) {
   // If the queue is non-empty, always queue the daragram.  This ensures that
@@ -27,6 +33,9 @@
     QuicMemSliceSpan span(&datagram);
     MessageResult result = session_->SendMessage(span);
     if (result.status != MESSAGE_STATUS_BLOCKED) {
+      if (observer_) {
+        observer_->OnDatagramProcessed(result.status);
+      }
       return result.status;
     }
   }
@@ -46,6 +55,9 @@
   MessageResult result = session_->SendMessage(span);
   if (result.status != MESSAGE_STATUS_BLOCKED) {
     queue_.pop_front();
+    if (observer_) {
+      observer_->OnDatagramProcessed(result.status);
+    }
   }
   return result.status;
 }
@@ -80,6 +92,9 @@
   QuicTime now = clock_->ApproximateNow();
   while (!queue_.empty() && queue_.front().expiry <= now) {
     queue_.pop_front();
+    if (observer_) {
+      observer_->OnDatagramProcessed(absl::nullopt);
+    }
   }
 }
 
diff --git a/quic/core/quic_datagram_queue.h b/quic/core/quic_datagram_queue.h
index 0952402..e8f2815 100644
--- a/quic/core/quic_datagram_queue.h
+++ b/quic/core/quic_datagram_queue.h
@@ -5,6 +5,8 @@
 #ifndef QUICHE_QUIC_CORE_QUIC_DATAGRAM_QUEUE_H_
 #define QUICHE_QUIC_CORE_QUIC_DATAGRAM_QUEUE_H_
 
+#include <memory>
+
 #include "absl/types/optional.h"
 #include "net/third_party/quiche/src/quic/core/quic_circular_deque.h"
 #include "net/third_party/quiche/src/quic/core/quic_time.h"
@@ -20,9 +22,26 @@
 // amount of time, and deleted after that time passes.
 class QUIC_EXPORT_PRIVATE QuicDatagramQueue {
  public:
+  // An interface used to monitor events on the associated `QuicDatagramQueue`.
+  class QUIC_EXPORT_PRIVATE Observer {
+   public:
+    virtual ~Observer() = default;
+
+    // Called when a datagram in the associated queue is sent or discarded.
+    // Identity information for the datagram is not given, because the sending
+    // and discarding order is always first-in-first-out.
+    // This function is called synchronously in `QuicDatagramQueue` methods.
+    // `status` is nullopt when the datagram is dropped due to being in the
+    // queue for too long.
+    virtual void OnDatagramProcessed(absl::optional<MessageStatus> status) = 0;
+  };
+
   // |session| is not owned and must outlive this object.
   explicit QuicDatagramQueue(QuicSession* session);
 
+  // |session| is not owned and must outlive this object.
+  QuicDatagramQueue(QuicSession* session, std::unique_ptr<Observer> observer);
+
   // Adds the datagram to the end of the queue.  May send it immediately; if
   // not, MESSAGE_STATUS_BLOCKED is returned.
   MessageStatus SendOrQueueDatagram(QuicMemSlice datagram);
@@ -62,6 +81,7 @@
 
   QuicTime::Delta max_time_in_queue_ = QuicTime::Delta::Zero();
   QuicCircularDeque<Datagram> queue_;
+  std::unique_ptr<Observer> observer_;
 };
 
 }  // namespace quic
diff --git a/quic/core/quic_datagram_queue_test.cc b/quic/core/quic_datagram_queue_test.cc
index 0e6e3de..e437840 100644
--- a/quic/core/quic_datagram_queue_test.cc
+++ b/quic/core/quic_datagram_queue_test.cc
@@ -4,6 +4,8 @@
 
 #include "net/third_party/quiche/src/quic/core/quic_datagram_queue.h"
 
+#include <vector>
+
 #include "absl/strings/string_view.h"
 #include "absl/types/optional.h"
 #include "net/third_party/quiche/src/quic/core/crypto/null_encrypter.h"
@@ -11,6 +13,7 @@
 #include "net/third_party/quiche/src/quic/core/quic_time.h"
 #include "net/third_party/quiche/src/quic/core/quic_types.h"
 #include "net/third_party/quiche/src/quic/platform/api/quic_mem_slice.h"
+#include "net/third_party/quiche/src/quic/platform/api/quic_reference_counted.h"
 #include "net/third_party/quiche/src/quic/platform/api/quic_test.h"
 #include "net/third_party/quiche/src/quic/test_tools/quic_test_utils.h"
 
@@ -30,20 +33,43 @@
   bool encryption_established() const override { return true; }
 };
 
-class QuicDatagramQueueTest : public QuicTest {
+class QuicDatagramQueueObserver final : public QuicDatagramQueue::Observer {
  public:
-  QuicDatagramQueueTest()
+  class Context : public QuicReferenceCounted {
+   public:
+    std::vector<absl::optional<MessageStatus>> statuses;
+  };
+
+  QuicDatagramQueueObserver() : context_(new Context()) {}
+  QuicDatagramQueueObserver(const QuicDatagramQueueObserver&) = delete;
+  QuicDatagramQueueObserver& operator=(const QuicDatagramQueueObserver&) =
+      delete;
+
+  void OnDatagramProcessed(absl::optional<MessageStatus> status) override {
+    context_->statuses.push_back(std::move(status));
+  }
+
+  const QuicReferenceCountedPointer<Context>& context() { return context_; }
+
+ private:
+  QuicReferenceCountedPointer<Context> context_;
+};
+
+class QuicDatagramQueueTestBase : public QuicTest {
+ protected:
+  QuicDatagramQueueTestBase()
       : connection_(new MockQuicConnection(&helper_,
                                            &alarm_factory_,
                                            Perspective::IS_CLIENT)),
-        session_(connection_),
-        queue_(&session_) {
+        session_(connection_) {
     session_.SetCryptoStream(new EstablishedCryptoStream(&session_));
     connection_->SetEncrypter(
         ENCRYPTION_FORWARD_SECURE,
         std::make_unique<NullEncrypter>(connection_->perspective()));
   }
 
+  ~QuicDatagramQueueTestBase() = default;
+
   QuicMemSlice CreateMemSlice(absl::string_view data) {
     QuicUniqueBufferPtr buffer =
         MakeUniqueBuffer(helper_.GetStreamSendBufferAllocator(), data.size());
@@ -51,11 +77,17 @@
     return QuicMemSlice(std::move(buffer), data.size());
   }
 
- protected:
   MockQuicConnectionHelper helper_;
   MockAlarmFactory alarm_factory_;
   MockQuicConnection* connection_;  // Owned by |session_|.
   MockQuicSession session_;
+};
+
+class QuicDatagramQueueTest : public QuicDatagramQueueTestBase {
+ public:
+  QuicDatagramQueueTest() : queue_(&session_) {}
+
+ protected:
   QuicDatagramQueue queue_;
 };
 
@@ -167,6 +199,99 @@
   EXPECT_EQ(0u, queue_.SendDatagrams());
 }
 
+class QuicDatagramQueueWithObserverTest : public QuicDatagramQueueTestBase {
+ public:
+  QuicDatagramQueueWithObserverTest()
+      : observer_(std::make_unique<QuicDatagramQueueObserver>()),
+        context_(observer_->context()),
+        queue_(&session_, std::move(observer_)) {}
+
+ protected:
+  // This is moved out immediately.
+  std::unique_ptr<QuicDatagramQueueObserver> observer_;
+
+  QuicReferenceCountedPointer<QuicDatagramQueueObserver::Context> context_;
+  QuicDatagramQueue queue_;
+};
+
+TEST_F(QuicDatagramQueueWithObserverTest, ObserveSuccessImmediately) {
+  EXPECT_TRUE(context_->statuses.empty());
+
+  EXPECT_CALL(*connection_, SendMessage(_, _, _))
+      .WillOnce(Return(MESSAGE_STATUS_SUCCESS));
+
+  EXPECT_EQ(MESSAGE_STATUS_SUCCESS,
+            queue_.SendOrQueueDatagram(CreateMemSlice("a")));
+
+  EXPECT_THAT(context_->statuses, ElementsAre(MESSAGE_STATUS_SUCCESS));
+}
+
+TEST_F(QuicDatagramQueueWithObserverTest, ObserveFailureImmediately) {
+  EXPECT_TRUE(context_->statuses.empty());
+
+  EXPECT_CALL(*connection_, SendMessage(_, _, _))
+      .WillOnce(Return(MESSAGE_STATUS_TOO_LARGE));
+
+  EXPECT_EQ(MESSAGE_STATUS_TOO_LARGE,
+            queue_.SendOrQueueDatagram(CreateMemSlice("a")));
+
+  EXPECT_THAT(context_->statuses, ElementsAre(MESSAGE_STATUS_TOO_LARGE));
+}
+
+TEST_F(QuicDatagramQueueWithObserverTest, BlockingShouldNotBeObserved) {
+  EXPECT_TRUE(context_->statuses.empty());
+
+  EXPECT_CALL(*connection_, SendMessage(_, _, _))
+      .WillRepeatedly(Return(MESSAGE_STATUS_BLOCKED));
+
+  EXPECT_EQ(MESSAGE_STATUS_BLOCKED,
+            queue_.SendOrQueueDatagram(CreateMemSlice("a")));
+  EXPECT_EQ(0u, queue_.SendDatagrams());
+
+  EXPECT_TRUE(context_->statuses.empty());
+}
+
+TEST_F(QuicDatagramQueueWithObserverTest, ObserveSuccessAfterBuffering) {
+  EXPECT_TRUE(context_->statuses.empty());
+
+  EXPECT_CALL(*connection_, SendMessage(_, _, _))
+      .WillOnce(Return(MESSAGE_STATUS_BLOCKED));
+
+  EXPECT_EQ(MESSAGE_STATUS_BLOCKED,
+            queue_.SendOrQueueDatagram(CreateMemSlice("a")));
+
+  EXPECT_TRUE(context_->statuses.empty());
+
+  EXPECT_CALL(*connection_, SendMessage(_, _, _))
+      .WillOnce(Return(MESSAGE_STATUS_SUCCESS));
+
+  EXPECT_EQ(1u, queue_.SendDatagrams());
+  EXPECT_THAT(context_->statuses, ElementsAre(MESSAGE_STATUS_SUCCESS));
+}
+
+TEST_F(QuicDatagramQueueWithObserverTest, ObserveExpiry) {
+  constexpr QuicTime::Delta expiry = QuicTime::Delta::FromMilliseconds(100);
+  queue_.SetMaxTimeInQueue(expiry);
+
+  EXPECT_TRUE(context_->statuses.empty());
+
+  EXPECT_CALL(*connection_, SendMessage(_, _, _))
+      .WillOnce(Return(MESSAGE_STATUS_BLOCKED));
+
+  EXPECT_EQ(MESSAGE_STATUS_BLOCKED,
+            queue_.SendOrQueueDatagram(CreateMemSlice("a")));
+
+  EXPECT_TRUE(context_->statuses.empty());
+
+  EXPECT_CALL(*connection_, SendMessage(_, _, _)).Times(0);
+  helper_.AdvanceTime(100 * expiry);
+
+  EXPECT_TRUE(context_->statuses.empty());
+
+  EXPECT_EQ(0u, queue_.SendDatagrams());
+  EXPECT_THAT(context_->statuses, ElementsAre(absl::nullopt));
+}
+
 }  // namespace
 }  // namespace test
 }  // namespace quic
diff --git a/quic/core/quic_session.cc b/quic/core/quic_session.cc
index 3a06af3..91aca8f 100644
--- a/quic/core/quic_session.cc
+++ b/quic/core/quic_session.cc
@@ -57,6 +57,20 @@
     const QuicConfig& config,
     const ParsedQuicVersionVector& supported_versions,
     QuicStreamCount num_expected_unidirectional_static_streams)
+    : QuicSession(connection,
+                  owner,
+                  config,
+                  supported_versions,
+                  num_expected_unidirectional_static_streams,
+                  nullptr) {}
+
+QuicSession::QuicSession(
+    QuicConnection* connection,
+    Visitor* owner,
+    const QuicConfig& config,
+    const ParsedQuicVersionVector& supported_versions,
+    QuicStreamCount num_expected_unidirectional_static_streams,
+    std::unique_ptr<QuicDatagramQueue::Observer> datagram_observer)
     : connection_(connection),
       perspective_(connection->perspective()),
       visitor_(owner),
@@ -94,7 +108,7 @@
       transport_goaway_received_(false),
       control_frame_manager_(this),
       last_message_id_(0),
-      datagram_queue_(this),
+      datagram_queue_(this, std::move(datagram_observer)),
       closed_streams_clean_up_alarm_(nullptr),
       supported_versions_(supported_versions),
       use_http2_priority_write_scheduler_(false),
diff --git a/quic/core/quic_session.h b/quic/core/quic_session.h
index e454892..a36040e 100644
--- a/quic/core/quic_session.h
+++ b/quic/core/quic_session.h
@@ -88,6 +88,12 @@
               const QuicConfig& config,
               const ParsedQuicVersionVector& supported_versions,
               QuicStreamCount num_expected_unidirectional_static_streams);
+  QuicSession(QuicConnection* connection,
+              Visitor* owner,
+              const QuicConfig& config,
+              const ParsedQuicVersionVector& supported_versions,
+              QuicStreamCount num_expected_unidirectional_static_streams,
+              std::unique_ptr<QuicDatagramQueue::Observer> datagram_observer);
   QuicSession(const QuicSession&) = delete;
   QuicSession& operator=(const QuicSession&) = delete;
 
diff --git a/quic/quic_transport/quic_transport_client_session.cc b/quic/quic_transport/quic_transport_client_session.cc
index f38d6b3..66f993a 100644
--- a/quic/quic_transport/quic_transport_client_session.cc
+++ b/quic/quic_transport/quic_transport_client_session.cc
@@ -34,12 +34,14 @@
     const GURL& url,
     QuicCryptoClientConfig* crypto_config,
     url::Origin origin,
-    ClientVisitor* visitor)
+    ClientVisitor* visitor,
+    std::unique_ptr<QuicDatagramQueue::Observer> datagram_observer)
     : QuicSession(connection,
                   owner,
                   config,
                   supported_versions,
-                  /*num_expected_unidirectional_static_streams*/ 0),
+                  /*num_expected_unidirectional_static_streams*/ 0,
+                  std::move(datagram_observer)),
       url_(url),
       origin_(origin),
       visitor_(visitor) {
diff --git a/quic/quic_transport/quic_transport_client_session.h b/quic/quic_transport/quic_transport_client_session.h
index c4f417f..b7687c0 100644
--- a/quic/quic_transport/quic_transport_client_session.h
+++ b/quic/quic_transport/quic_transport_client_session.h
@@ -16,6 +16,7 @@
 #include "net/third_party/quiche/src/quic/core/quic_connection.h"
 #include "net/third_party/quiche/src/quic/core/quic_crypto_client_stream.h"
 #include "net/third_party/quiche/src/quic/core/quic_crypto_stream.h"
+#include "net/third_party/quiche/src/quic/core/quic_datagram_queue.h"
 #include "net/third_party/quiche/src/quic/core/quic_server_id.h"
 #include "net/third_party/quiche/src/quic/core/quic_session.h"
 #include "net/third_party/quiche/src/quic/core/quic_stream.h"
@@ -56,14 +57,16 @@
     virtual void OnCanCreateNewOutgoingUnidirectionalStream() = 0;
   };
 
-  QuicTransportClientSession(QuicConnection* connection,
-                             Visitor* owner,
-                             const QuicConfig& config,
-                             const ParsedQuicVersionVector& supported_versions,
-                             const GURL& url,
-                             QuicCryptoClientConfig* crypto_config,
-                             url::Origin origin,
-                             ClientVisitor* visitor);
+  QuicTransportClientSession(
+      QuicConnection* connection,
+      Visitor* owner,
+      const QuicConfig& config,
+      const ParsedQuicVersionVector& supported_versions,
+      const GURL& url,
+      QuicCryptoClientConfig* crypto_config,
+      url::Origin origin,
+      ClientVisitor* visitor,
+      std::unique_ptr<QuicDatagramQueue::Observer> datagram_observer);
 
   std::vector<std::string> GetAlpnsToOffer() const override {
     return std::vector<std::string>({QuicTransportAlpn()});
diff --git a/quic/quic_transport/quic_transport_client_session_test.cc b/quic/quic_transport/quic_transport_client_session_test.cc
index 129c87e..b311b52 100644
--- a/quic/quic_transport/quic_transport_client_session_test.cc
+++ b/quic/quic_transport/quic_transport_client_session_test.cc
@@ -65,7 +65,7 @@
     session_ = std::make_unique<QuicTransportClientSession>(
         &connection_, nullptr, DefaultQuicConfig(), GetVersions(),
         GURL("quic-transport://test.example.com:50000" + url_suffix),
-        &crypto_config_, origin, &visitor_);
+        &crypto_config_, origin, &visitor_, /*datagram_observer=*/nullptr);
     session_->Initialize();
     crypto_stream_ = static_cast<QuicCryptoClientStream*>(
         session_->GetMutableCryptoStream());
diff --git a/quic/quic_transport/quic_transport_integration_test.cc b/quic/quic_transport/quic_transport_integration_test.cc
index 1965c6d..99c20b2 100644
--- a/quic/quic_transport/quic_transport_integration_test.cc
+++ b/quic/quic_transport/quic_transport_integration_test.cc
@@ -90,7 +90,8 @@
                  GURL("quic-transport://test.example.com:50000" + path),
                  &crypto_config_,
                  origin,
-                 &visitor_) {
+                 &visitor_,
+                 /*datagram_observer=*/nullptr) {
     session_.Initialize();
   }