This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 868ac1713963 [SPARK-57664][CONNECT] Make dropped pipeline events
observable in PipelineEventSender
868ac1713963 is described below
commit 868ac17139638e7c6a6ed04f74ea576651d21ad0
Author: YangJie <[email protected]>
AuthorDate: Tue Jul 7 12:20:12 2026 +0800
[SPARK-57664][CONNECT] Make dropped pipeline events observable in
PipelineEventSender
### What changes were proposed in this pull request?
`PipelineEventSender` streams pipeline events to a Connect client through a
single background thread backed by a bounded queue (sized by the event-queue
capacity conf). When that queue is full it intentionally drops non-terminal
`FlowProgress` and other non-`RunProgress` events to avoid blocking execution;
terminal `FlowProgress` events and all `RunProgress` events are always
enqueued. Until now those drops were completely silent.
This PR makes them observable:
- a running count of dropped events, exposed as `numDroppedEvents`;
- a warning logged when an event is dropped, throttled to at most once per
minute (the first drop always logs) so a persistently full queue does not flood
the logs;
- a summary warning at shutdown reporting the total, so drops that were
suppressed by the throttle window are still surfaced.
The throttling follows the approach `AsyncEventQueue` already uses for the
same "queue full, drop events" situation.
### Why are the changes needed?
A dropped event is lost progress reporting to the client. Previously a drop
produced no log line and no counter, so an operator who noticed gaps in
pipeline progress had no signal that events were being discarded, let alone how
many.
### Does this PR introduce _any_ user-facing change?
No. The counter is internal; the only outward change is the additional WARN
log lines emitted when events are dropped.
### How was this patch tested?
New unit tests in `PipelineEventSenderSuite`:
- a sender that never overflows its queue reports zero dropped events;
- events dropped at capacity are counted (`numDroppedEvents`);
- the per-drop warning is throttled (logged once for two in-window drops)
and the shutdown summary is emitted;
- the warning is re-logged once the throttle interval has elapsed.
The drop scenarios use a `CountDownLatch` handshake to park the worker
deterministically rather than relying on timing. The existing capacity test
continues to cover the enqueue/drop routing.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56742 from LuciferYang/sdp-event-drop-observability.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
---
.../connect/pipelines/PipelineEventSender.scala | 56 ++++++++++-
.../pipelines/PipelineEventSenderSuite.scala | 109 +++++++++++++++++++++
2 files changed, 164 insertions(+), 1 deletion(-)
diff --git
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelineEventSender.scala
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelineEventSender.scala
index 79dd07b6a658..e5e4a8ee6e77 100644
---
a/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelineEventSender.scala
+++
b/sql/connect/server/src/main/scala/org/apache/spark/sql/connect/pipelines/PipelineEventSender.scala
@@ -18,7 +18,7 @@
package org.apache.spark.sql.connect.pipelines
import java.util.concurrent.ThreadPoolExecutor
-import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}
import scala.util.control.NonFatal
@@ -62,6 +62,17 @@ class PipelineEventSender(
*/
private val isShutdown = new AtomicBoolean(false)
+ // Running total of non-terminal events dropped because the queue was full.
+ private val droppedEventCount = new AtomicLong(0L)
+
+ // Timestamp (ms) of the last dropped-event warning, used to throttle them.
Only read/written in
+ // `recordDroppedEvent`, which runs inside the synchronized `sendEvent`, so
a plain var suffices.
+ private var lastDropWarningTimestamp = 0L
+
+ // Overridable so tests can drive the throttle-reset path without waiting on
the wall clock.
+ protected def droppedEventLogIntervalMs: Long =
+ PipelineEventSender.DROPPED_EVENT_LOG_INTERVAL_MS
+
/**
* Send an event async by submitting it to the executor, if the sender is
not shut down.
* Otherwise, throws an IllegalStateException, to raise awareness of the
shutdown state.
@@ -85,12 +96,39 @@ class PipelineEventSender(
}
}
})
+ } else {
+ recordDroppedEvent(event)
}
} else {
throw
IllegalStateErrors.eventSendAfterShutdown(sessionHolder.key.toString)
}
}
+ /**
+ * Record a non-terminal event that was dropped because the queue was at
capacity. Always bumps
+ * the counter so the running total reported in the warnings stays accurate,
and logs a warning
+ * at most once per [[droppedEventLogIntervalMs]] (the first drop always
logs).
+ */
+ private def recordDroppedEvent(event: PipelineEvent): Unit = {
+ val totalDropped = droppedEventCount.incrementAndGet()
+ val now = System.currentTimeMillis()
+ if (now - lastDropWarningTimestamp >= droppedEventLogIntervalMs) {
+ lastDropWarningTimestamp = now
+ logWarning(
+ log"Dropped pipeline event for session " +
+ log"${MDC(LogKeys.SESSION_ID, sessionHolder.sessionId)} because the
event queue is " +
+ log"full (capacity ${MDC(LogKeys.MAX_SIZE, queueCapacity)}); " +
+ log"${MDC(LogKeys.NUM_EVENTS, totalDropped)} non-terminal event(s)
dropped so far. " +
+ log"Most recent dropped event: ${MDC(LogKeys.MESSAGE,
event.message)}")
+ }
+ }
+
+ /**
+ * Total number of non-terminal events dropped because the queue was full.
Exposed for tests;
+ * production observability is the warning logs emitted on drop and at
shutdown.
+ */
+ private[connect] def numDroppedEvents: Long = droppedEventCount.get()
+
private def shouldEnqueueEvent(event: PipelineEvent): Boolean = {
event.details match {
case _: RunProgress =>
@@ -127,6 +165,16 @@ class PipelineEventSender(
log"${MDC(LogKeys.SESSION_ID, sessionHolder.sessionId)} failed to
terminate")
executor.shutdownNow()
}
+ // Summarize total drops at shutdown so drops suppressed by the throttle
window are still
+ // surfaced.
+ val totalDropped = droppedEventCount.get()
+ if (totalDropped > 0) {
+ logWarning(
+ log"Pipeline event sender for session " +
+ log"${MDC(LogKeys.SESSION_ID, sessionHolder.sessionId)} dropped a
total of " +
+ log"${MDC(LogKeys.NUM_EVENTS, totalDropped)} non-terminal event(s)
because the " +
+ log"event queue (capacity ${MDC(LogKeys.MAX_SIZE, queueCapacity)})
was full.")
+ }
logInfo(
log"Pipeline event sender shutdown completed for session " +
log"${MDC(LogKeys.SESSION_ID, sessionHolder.sessionId)}")
@@ -174,3 +222,9 @@ class PipelineEventSender(
protoEventBuilder.build()
}
}
+
+object PipelineEventSender {
+ // Minimum interval between dropped-event warnings, to avoid flooding the
logs when the queue
+ // stays full. Mirrors AsyncEventQueue.LOGGING_INTERVAL.
+ private val DROPPED_EVENT_LOG_INTERVAL_MS = 60L * 1000
+}
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PipelineEventSenderSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PipelineEventSenderSuite.scala
index d6942132b908..22f620d9d07a 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PipelineEventSenderSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PipelineEventSenderSuite.scala
@@ -18,8 +18,10 @@
package org.apache.spark.sql.connect.pipelines
import java.sql.Timestamp
+import java.util.concurrent.{CountDownLatch, TimeUnit}
import io.grpc.stub.StreamObserver
+import org.apache.logging.log4j.Level
import org.mockito.{ArgumentCaptor, Mockito}
import org.mockito.Mockito._
import org.scalatestplus.mockito.MockitoSugar
@@ -89,6 +91,9 @@ class PipelineEventSenderSuite extends
SparkDeclarativePipelinesServerTest with
val pipelineEvent = response.getPipelineEventResult.getEvent
assert(pipelineEvent.getMessage == "Test message")
+
+ // A healthy sender that never overflows its queue must report zero
dropped events.
+ assert(eventSender.numDroppedEvents == 0)
} finally {
eventSender.shutdown()
}
@@ -230,4 +235,108 @@ class PipelineEventSenderSuite extends
SparkDeclarativePipelinesServerTest with
eventSender.shutdown()
}
}
+
+ test("PipelineEventSender logs a warning when it drops events due to a full
queue") {
+ val (mockObserver, mockSessionHolder) = createMockSetup(queueSize = "1")
+
+ // `workerStarted` fires once the background thread has dequeued and begun
processing the
+ // first event; `release` keeps that thread parked so the queue fills
deterministically
+ // (no timing assumptions) while we submit the events that must be dropped.
+ val workerStarted = new CountDownLatch(1)
+ val release = new CountDownLatch(1)
+ val eventSender = new PipelineEventSender(mockObserver, mockSessionHolder)
{
+ override def sendEventToClient(event: PipelineEvent): Unit = {
+ workerStarted.countDown()
+ release.await()
+ super.sendEventToClient(event)
+ }
+ }
+
+ val logAppender = new LogAppender("dropped pipeline event warning")
+ try {
+ withLogAppender(logAppender, level = Some(Level.WARN)) {
+ // Occupy the single worker thread and wait until it is actually
parked.
+ eventSender.sendEvent(
+ createTestEvent(id = "started", details =
FlowProgress(FlowStatus.STARTING)))
+ assert(workerStarted.await(10, TimeUnit.SECONDS), "worker thread did
not start")
+
+ // Fill the single queue slot.
+ eventSender.sendEvent(
+ createTestEvent(id = "queued", details =
FlowProgress(FlowStatus.RUNNING)))
+
+ // These two non-terminal events have nowhere to go and must be
dropped.
+ eventSender.sendEvent(
+ createTestEvent(id = "dropped-1", details =
FlowProgress(FlowStatus.RUNNING)))
+ eventSender.sendEvent(
+ createTestEvent(id = "dropped-2", details =
FlowProgress(FlowStatus.RUNNING)))
+
+ release.countDown()
+ eventSender.shutdown() // drains the started + queued events and logs
the drop summary
+ }
+
+ assert(eventSender.numDroppedEvents == 2)
+
+ val warnings = logAppender.loggingEvents
+ .filter(_.getLevel == Level.WARN)
+ .map(_.getMessage.getFormattedMessage)
+ // The two drops are submitted back-to-back, far inside the default 60s
throttle window, so
+ // the per-drop warning fires exactly once (first logs, second
suppressed) and the running
+ // total is logged at shutdown. The assertions match on substrings so
they stay robust to
+ // the exact WARN wording.
+ assert(warnings.count(_.contains("Dropped pipeline event for session"))
== 1)
+ assert(warnings.exists(_.contains("dropped a total of")))
+ } finally {
+ release.countDown()
+ eventSender.shutdown()
+ }
+ }
+
+ test("PipelineEventSender re-logs a drop warning once the throttle interval
has elapsed") {
+ val (mockObserver, mockSessionHolder) = createMockSetup(queueSize = "1")
+
+ val workerStarted = new CountDownLatch(1)
+ val release = new CountDownLatch(1)
+ // A negative interval makes the throttle treat every drop as past the
window, so each drop
+ // re-logs. This exercises the re-log branch deterministically,
independent of the wall clock
+ // (a zero interval would instead depend on currentTimeMillis() being
monotonic across drops).
+ val eventSender = new PipelineEventSender(mockObserver, mockSessionHolder)
{
+ override protected def droppedEventLogIntervalMs: Long = Long.MinValue
+ override def sendEventToClient(event: PipelineEvent): Unit = {
+ workerStarted.countDown()
+ release.await()
+ super.sendEventToClient(event)
+ }
+ }
+
+ val logAppender = new LogAppender("throttle reset warning")
+ try {
+ withLogAppender(logAppender, level = Some(Level.WARN)) {
+ eventSender.sendEvent(
+ createTestEvent(id = "started", details =
FlowProgress(FlowStatus.STARTING)))
+ assert(workerStarted.await(10, TimeUnit.SECONDS), "worker thread did
not start")
+
+ eventSender.sendEvent(
+ createTestEvent(id = "queued", details =
FlowProgress(FlowStatus.RUNNING)))
+ eventSender.sendEvent(
+ createTestEvent(id = "dropped-1", details =
FlowProgress(FlowStatus.RUNNING)))
+ eventSender.sendEvent(
+ createTestEvent(id = "dropped-2", details =
FlowProgress(FlowStatus.RUNNING)))
+
+ release.countDown()
+ eventSender.shutdown()
+ }
+
+ assert(eventSender.numDroppedEvents == 2)
+
+ val warnings = logAppender.loggingEvents
+ .filter(_.getLevel == Level.WARN)
+ .map(_.getMessage.getFormattedMessage)
+ // With the throttle disabled, both drops re-arm the warning, so the
per-drop message is
+ // logged once per drop rather than being suppressed.
+ assert(warnings.count(_.contains("Dropped pipeline event for session"))
== 2)
+ } finally {
+ release.countDown()
+ eventSender.shutdown()
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
