chore: enable diagnostics by default for pinned branch

Co-Authored-By: gl_anatolii.yatsuk <gl_anatolii.yatsuk@airbyte.io>

Author: devin-ai-integration[bot]

airbyte_cdk/entrypoint.py

@@ -53,7 +53,6 @@
 VALID_URL_SCHEMES = ["https"]
 CLOUD_DEPLOYMENT_MODE = "cloud"
 _HAS_LOGGED_FOR_SERIALIZATION_ERROR = False
-_DEADLOCK_DIAGNOSTICS_ENV = "AIRBYTE_CDK_DEADLOCK_DIAGNOSTICS"
 
 
 class _DeadlockDiagnostics:
@@ -534,13 +533,8 @@ def _emit_queued_messages(self, source: Source) -> Iterable[AirbyteMessage]:
 def launch(source: Source, args: List[str]) -> None:
     source_entrypoint = AirbyteEntrypoint(source)
     parsed_args = source_entrypoint.parse_args(args)
-    diagnostics = (
-        _DeadlockDiagnostics()
-        if os.environ.get(_DEADLOCK_DIAGNOSTICS_ENV, "").lower() == "true"
-        else None
-    )
-    if diagnostics:
-        diagnostics.start()
+    diagnostics = _DeadlockDiagnostics()
+    diagnostics.start()
 
     # temporarily removes the PrintBuffer because we're seeing weird print behavior for concurrent syncs
     # Refer to: https://github.com/airbytehq/oncall/issues/6235
@@ -550,18 +544,14 @@ def launch(source: Source, args: List[str]) -> None:
                 # simply printing is creating issues for concurrent CDK as Python uses different two instructions to print: one for the message and
                 # the other for the break line. Adding `\n` to the message ensure that both are printed at the same time
                 data = f"{message}\n"
-                if diagnostics:
-                    diagnostics.mark_print_started()
+                diagnostics.mark_print_started()
                 try:
                     print(data, end="")
                 finally:
-                    if diagnostics:
-                        diagnostics.mark_print_finished()
-                if diagnostics:
-                    diagnostics.record_message(data)
+                    diagnostics.mark_print_finished()
+                diagnostics.record_message(data)
     finally:
-        if diagnostics:
-            diagnostics.stop()
+        diagnostics.stop()
 
 
 def _init_internal_request_filter() -> None:

unit_tests/test_entrypoint.py

@@ -994,10 +994,9 @@ def test_deadlock_diagnostics_heartbeat_reports_queue_stats():
     assert "queue_size=1 queue_full=True" in emitted[0].decode()
 
 
-def test_launch_starts_deadlock_diagnostics_when_enabled(mocker):
+def test_launch_starts_deadlock_diagnostics(mocker):
     diagnostics = MagicMock()
     mocker.patch.object(entrypoint_module, "_DeadlockDiagnostics", return_value=diagnostics)
-    mocker.patch.dict(os.environ, {entrypoint_module._DEADLOCK_DIAGNOSTICS_ENV: "true"})
     mocker.patch.object(AirbyteEntrypoint, "parse_args", return_value=Namespace(command="spec"))
     mocker.patch.object(
         AirbyteEntrypoint,
@@ -1018,25 +1017,3 @@ def test_launch_starts_deadlock_diagnostics_when_enabled(mocker):
     diagnostics.mark_print_finished.assert_called_once_with()
     diagnostics.record_message.assert_called_once()
     diagnostics.stop.assert_called_once_with()
-
-
-def test_launch_does_not_start_deadlock_diagnostics_by_default(mocker):
-    diagnostics = MagicMock()
-    mocker.patch.object(entrypoint_module, "_DeadlockDiagnostics", return_value=diagnostics)
-    mocker.patch.dict(os.environ, {}, clear=True)
-    mocker.patch.object(AirbyteEntrypoint, "parse_args", return_value=Namespace(command="spec"))
-    mocker.patch.object(
-        AirbyteEntrypoint,
-        "run",
-        return_value=iter(
-            [
-                AirbyteMessage(
-                    type=Type.SPEC, spec=ConnectorSpecification(connectionSpecification={})
-                )
-            ]
-        ),
-    )
-
-    entrypoint_module.launch(MockSource(), ["spec"])
-
-    entrypoint_module._DeadlockDiagnostics.assert_not_called()