eugenegujing commented on code in PR #6724:
URL: https://github.com/apache/texera/pull/6724#discussion_r3779953039
##########
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:
Added in 1a65167ba49f280df7cfeaf195adb6c87a3689b7. The matrix now enumerates
a channel's first element by kind (`DataElement` / `ECMElement` /
`DCMElement`), plus dedicated ECM-first tests for both PAUSE and BACKPRESSURE
carrying a 2s timeout, so a regression surfaces as a timeout rather than a
hang. Reinstating the born-disabled gate turns them red.
--
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]
