Stop throwing an error in an attempt to stop an already failed statement

CHANGELOG.md

@@ -7,6 +7,7 @@ All notable changes to this dbapi driver will be documented in this file.
 ### Fixed
 
 - Type conversion methods across `types.py` now consistently raise dbapi-mandated exceptions (`DataError`, `InterfaceError`) rather than a bare builtin (`ValueError`, `decimal.InvalidOperation`) for problems with a Flink response value, a Python value that can't be represented as a Flink SQL literal, or a converter misconfigured with the wrong column type. (#204)
+- `stop_statement(wait_for_stopped=True)` (the default) on a statement that had already reached FAILED on its own -- before the stop was ever requested -- no longer raises `OperationalError`. The blocking wait mistook "the statement is currently FAILED" for "the statement transitioned to FAILED while we were waiting for it to stop," when the former is really the same "already terminal, nothing to stop" success the `Statement`-object short-circuit already returns for a cached terminal statement; only a transition *into* FAILED partway through the wait is a genuine failure to stop cleanly. (#203)
 
 ### Changed
 

src/confluent_sql/connection.py

@@ -1255,9 +1255,13 @@ def stop_statement(
         phase RUNNING.
 
         Args:
-            statement: The name of the statement to stop, or a Statement object. A Statement
-                already in a terminal phase (STOPPED/COMPLETED/FAILED/DELETED) is returned
-                unchanged without contacting the server.
+            statement: The name of the statement to stop, or a Statement object. Either form
+                converges on the same outcome for a statement already in a terminal phase
+                (STOPPED/COMPLETED/FAILED/DELETED): a `Statement` object is returned unchanged
+                without contacting the server; a name is still PATCHed (the server accepts this
+                unconditionally, even against an already-terminal statement -- there is no
+                rejection to handle), and the terminal state the PATCH response already reflects
+                is returned as success, not raised as an error.
             wait_for_stopped: If True (default), block and refresh-loop until the statement reaches
                 a terminal phase before returning -- normally STOPPED, but COMPLETED if a bounded
                 query happened to finish before the stop landed. If False, return as soon as the
@@ -1328,13 +1332,18 @@ def _wait_for_statement_stopped(self, statement: Statement, timeout: int) -> Sta
         so a GET issued immediately would almost certainly report the same non-terminal phase we
         already hold; we therefore sleep *first* and only fetch once wall-clock time has passed and
         the server state can actually have advanced. Uses exponential backoff with jitter to avoid
-        hammering the server. A statement that ends in any terminal phase other than FAILED (e.g. a
-        bounded query that COMPLETED before the stop landed) is returned, since the caller's intent
-        -- the statement is no longer running -- is satisfied.
+        hammering the server. A statement that ends in any terminal phase, including one already
+        FAILED before this stop was ever requested (confirmed live, #203: the stop PATCH is
+        accepted unconditionally, echoing back whatever phase the statement already has -- there's
+        no separate "already terminal" rejection to special-case), is returned as success, since
+        the caller's intent -- the statement is no longer running -- is satisfied either way. Only
+        a transition *into* FAILED partway through the wait (observed by a poll after the initial,
+        already-terminal check above has been passed) is treated as a genuine failure to stop
+        cleanly and raises.
 
         Raises:
-            OperationalError: If the statement transitions to FAILED, or if STOPPED is not reached
-                within the timeout.
+            OperationalError: If the statement transitions to FAILED partway through waiting, or if
+                STOPPED is not reached within the timeout.
         """
 
         def raise_if_failed(candidate: Statement) -> None:
@@ -1344,8 +1353,9 @@ def raise_if_failed(candidate: Statement) -> None:
                     f"{candidate.status.get('detail', '')}"
                 )
 
-        # Evaluate the state we already hold first -- an already-terminal statement needs no fetch.
-        raise_if_failed(statement)
+        # Evaluate the state we already hold first -- any terminal phase, FAILED included, needs
+        # no fetch and is not an error here: it mirrors the Statement-object short-circuit's
+        # "already terminal, nothing to stop" success for a cached terminal statement.
         if statement.phase.is_terminal:
             return statement
 

tests/integration/test_connection.py

@@ -13,7 +13,7 @@
 
 import confluent_sql
 from confluent_sql.connection import Connection
-from confluent_sql.exceptions import StatementNotFoundError
+from confluent_sql.exceptions import OperationalError, StatementNotFoundError
 from confluent_sql.execution_mode import ExecutionMode
 from confluent_sql.statement import Statement
 
@@ -374,7 +374,12 @@ def test_stopping_already_stopped_statement_is_noop(
         test_table_name: str,
         cleaned_up_statement_name: str,
     ):
-        """Re-stopping an already-STOPPED statement returns it and does not raise."""
+        """Re-stopping an already-STOPPED statement returns it and does not raise, whether
+        re-stopped by Statement object (cached-state short-circuit, no server call) or by bare
+        name. The server's stop PATCH is idempotent -- confirmed live -- so the bare-name path
+        was never actually broken for this particular (already-STOPPED) case; see
+        test_stopping_statement_that_already_failed_returns_without_raising below for the
+        scenario #203 actually fixes."""
         cursor, _ = _start_running_streaming_statement(
             table_connection, test_table_name, cleaned_up_statement_name
         )
@@ -387,6 +392,33 @@ def test_stopping_already_stopped_statement_is_noop(
             # Passing the already-terminal Statement back short-circuits with no error.
             again = table_connection.stop_statement(stopped)
             assert again.is_stopped
+
+            # Re-stopping by bare name is likewise a no-op, not an error.
+            again_by_name = table_connection.stop_statement(cleaned_up_statement_name)
+            assert again_by_name.is_stopped
+        finally:
+            cursor.close()
+
+    def test_stopping_statement_that_already_failed_returns_without_raising(
+        self,
+        connection: Connection,
+        cleaned_up_statement_name: str,
+    ):
+        """The actual bug behind #203: a statement that reached FAILED on its own (not via
+        stop_statement) previously raised OperationalError out of the blocking wait instead of
+        being recognized as already terminal -- see _wait_for_statement_stopped's docstring.
+
+        `SELECT 1/0` fails fast enough (confirmed live: well under a second) that execute() itself
+        raises with "submission failed" -- Cursor.execute() sets cursor.statement before that
+        raise (cursor.py:251-264), so the FAILED statement is still available to stop by name."""
+        cursor = connection.cursor(mode=ExecutionMode.STREAMING_QUERY)
+        try:
+            with pytest.raises(OperationalError, match="submission failed"):
+                cursor.execute("SELECT 1/0", statement_name=cleaned_up_statement_name)
+            assert cursor.statement.is_failed
+
+            stopped = connection.stop_statement(cleaned_up_statement_name, wait_for_stopped=True)
+            assert stopped.is_failed
         finally:
             cursor.close()
 

tests/unit/test_connection_unit.py

@@ -400,6 +400,35 @@ def test_other_http_error_raises_operational_error(
             invalid_credential_connection.stop_statement("stmt-1", wait_for_stopped=False)
         assert exc_info.value.http_status_code == 500
 
+    def test_blocking_returns_without_polling_when_patch_already_failed(
+        self,
+        invalid_credential_connection: Connection,
+        statement_response_factory: StatementResponseFactory,
+        mocker,
+    ):
+        """A statement that reached FAILED on its own *before* stop_statement() was ever called
+        still returns cleanly, matching the Statement-object short-circuit's "already terminal,
+        nothing to stop" success -- not an error. This is the actual #203 bug: confirmed live
+        against the real Flink Statements API, the stop PATCH is accepted unconditionally (200 OK)
+        even against an already-FAILED statement, simply echoing back its current phase; there is
+        no server-side rejection to special-case. The bug was that the blocking wait's
+        transitioned-to-FAILED check ran before its already-terminal check, so it raised for a
+        statement that was already FAILED on the very first look, not one that failed *during* the
+        wait (that case is covered separately by test_blocking_failed_raises below)."""
+        request_mock = mocker.patch.object(
+            invalid_credential_connection._get_flink_client(), "request"
+        )
+        request_mock.return_value = _ok_response(
+            statement_response_factory(name="stmt-1", phase="FAILED", stopped=True)
+        )
+
+        result = invalid_credential_connection.stop_statement("stmt-1", wait_for_stopped=True)
+
+        assert result.is_failed
+        # Just the PATCH -- already terminal, so no follow-up GET poll is needed.
+        request_mock.assert_called_once()
+        assert request_mock.call_args.args[0] == "PATCH"
+
     def test_blocking_timeout_raises(
         self,
         invalid_credential_connection: Connection,