Yicong-Huang commented on code in PR #6724:
URL: https://github.com/apache/texera/pull/6724#discussion_r3764810946
##########
amber/src/main/python/core/models/internal_queue.py:
##########
@@ -77,11 +76,19 @@ def get(self) -> T:
def put(self, item: T) -> None:
if isinstance(item, InternalQueueElement):
if item.tag not in self._queue_ids:
- self._queue.add_sub_queue(item.tag, 1 if item.tag.is_control
else 2)
- self._queue_ids.add(item.tag)
- if isinstance(item, (DataElement, InternalMarker, ECMElement)):
- self._queue.put(item.tag, item)
- elif isinstance(item, DCMElement):
+ # registration must not interleave with
disable_data/enable_data
+ with self._lock:
+ if item.tag not in self._queue_ids:
+ self._queue.add_sub_queue(
+ item.tag, 1 if item.tag.is_control else 2
+ )
+ # while data is disabled, a new data sub-queue must
+ # start disabled too (before its first element is
+ # enqueued), or it would leak data during
pause/backpressure
+ if not item.tag.is_control and self._queue_state:
Review Comment:
ECMs ride data channels: `network_receiver.py:110` tags them with the data
channel, and `_send_ecm_to_data_channels` sends only to `not is_control`. So
this gate makes an ECM that is a channel's first message undeliverable rather
than delayed — `LinkedBlockingMultiQueue` serves only sub-queues where
`child.enabled` (`linked_blocking_multi_queue.py:223`). A paused worker's
reconfiguration command then never arrives and never acks, and the
coordinator's 30s `reconfigureWorkflow` await expires.
You were right and my round-2 CI read was wrong — I retract it. The Iceberg
`CatalogCommitConflicts` lines are WARN retries. The failing suite is
`ReconfigurationIntegrationSpec`, 3 tests on both runners, while main's last 10
runs of this workflow are green.
Of your two options I'd take the dequeue-side one. Register enabled as on
main, then refuse to hand out a `DataElement` while `_queue_state` is
non-empty, disabling that channel and stashing the element there. ECM delivery
stays identical to main, so reconfiguration is untouched. Confirmed too:
`pause_manager.py:70-79` does re-enable `ECM_PAUSE` channels before it checks
`_global_pauses`.
##########
amber/src/test/python/core/models/test_internal_queue.py:
##########
@@ -364,3 +365,297 @@ def test_it_can_disable_and_enable_a_single_data_channel(
queue.enable(data_channel)
assert queue.get() is blocked
assert queue.is_empty()
+
+ # Regression tests below: data channels whose sub-queue is created lazily
+ # (on the channel's first put) AFTER disable_data has been called must
+ # come up disabled — a paused or backpressured worker must not be able to
+ # dequeue data from them, and is_data_enabled() must not flip back to
+ # True just because a new channel delivered its first message.
+
+ def test_channel_registered_after_disable_comes_up_disabled(
+ self, queue, data_channel
+ ):
+ # the main regression: disable first, then the channel's FIRST put
+ queue.disable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ queue.put(self.data_element(data_channel))
+ assert not queue.is_data_enabled()
+ # the element stays queued but must not be dequeuable
+ assert queue.size_data() == 1
+ assert queue._queue.peek() is None
+
+ @pytest.mark.timeout(2)
+ def test_enable_data_releases_a_channel_registered_mid_disable(
+ self, queue, data_channel
+ ):
+ queue.disable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ data = self.data_element(data_channel)
+ queue.put(data)
+ assert queue.enable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ assert queue.is_data_enabled()
+ assert queue._queue.peek() is data
+ assert queue.get() is data
+ assert queue.is_empty()
+
+ @pytest.mark.timeout(2)
+ def test_channel_registered_under_stacked_disables_stays_disabled(
+ self, queue, data_channel
+ ):
+ queue.disable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ queue.disable_data(InternalQueue.DisableType.DISABLE_BY_BACKPRESSURE)
+ data = self.data_element(data_channel)
+ queue.put(data)
+ # releasing only one of the two reasons must not open the channel
+ assert not
queue.enable_data(InternalQueue.DisableType.DISABLE_BY_PAUSE)
+ assert not queue.is_data_enabled()
+ assert queue._queue.peek() is None
+ # releasing the remaining reason makes the element dequeuable
+ assert
queue.enable_data(InternalQueue.DisableType.DISABLE_BY_BACKPRESSURE)
+ assert queue.is_data_enabled()
+ assert queue.get() is data
+
+ @pytest.mark.timeout(2)
+ def test_control_channel_registered_mid_disable_is_never_blocked(
Review Comment:
This is the neighbouring case that would have caught it. The matrix
enumerates registration by *channel* kind — a data channel first carrying a
`DataElement`, a control channel mid-pause — but never by *element* kind. So a
data channel whose first element is an `ECMElement` goes untested, and that is
the cell CI fails on. The `ecm_element` helper is already there at line 89.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
