Nothing Special   »   [go: up one dir, main page]
Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

"Broken Pipe Error" should be caught and handled by Meltano #7047

Open
aaronsteers opened this issue Dec 2, 2022 · 2 comments
Open

"Broken Pipe Error" should be caught and handled by Meltano #7047

aaronsteers opened this issue Dec 2, 2022 · 2 comments
Labels
kind/Bug Something isn't working urgency/High valuestream/Meltano

Comments

@aaronsteers
Copy link
Contributor
aaronsteers commented Dec 2, 2022 •
edited
Loading

Raised by @Stephen Hazeltine in Slack. Approximately 6 pages of log text in the traceback (copy-pasted below). I think we should raise this up to a parent error like MeltanoBrokenPipeError or similar, that links to https://melta.no/....

Often this occurs when the tap fails unexpectedly, which breaks the STDOUT > STDIN 'pipe' between tap and target.

Details 

2022-12-02T18:56:51.919760Z [error    ] [Errno 32] Broken pipe         
╭─────────────────────────────── Traceback (most recent call last) ────────────────────────────────╮
│ /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_2/lib/python3.8/site-packages/m │
│ eltano/core/logging/output_logger.py:201 in redirect_logging                                     │
│                                                                                                  │
│   198 │   │   │   *ignore_errors,                                                                │
│   199 │   │   )                                                                                  │
│   200 │   │   try:                                                                               │
│ ❱ 201 │   │   │   yield                                                                          │
│   202 │   │   except ignored_errors:  # noqa: WPS329                                             │
│   203 │   │   │   raise                                                                          │
│   204 │   │   except Exception as err:                                                           │
│                                                                                                  │
│ ╭──────────────────────────────────── locals ─────────────────────────────────────╮              │
│ │            err = BrokenPipeError(32, 'Broken pipe')                             │              │
│ │  ignore_errors = (<class 'meltano.cli.utils.CliError'>,)                        │              │
│ │ ignored_errors = (                                                              │              │
│ │                  │   <class 'KeyboardInterrupt'>,                               │              │
│ │                  │   <class 'asyncio.exceptions.CancelledError'>,               │              │
│ │                  │   <class 'meltano.cli.utils.CliError'>                       │              │
│ │                  )                                                              │              │
│ │         logger = <RootLogger root (INFO)>                                       │              │
│ │           self = <meltano.core.logging.output_logger.Out object at 0x114357820> │              │
│ ╰─────────────────────────────────────────────────────────────────────────────────╯              │
│                                                                                                  │
│ /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_2/lib/python3.8/site-packages/m │
│ eltano/cli/elt.py:257 in _redirect_output                                                        │
│                                                                                                  │
│   254 │   with meltano_stdout.redirect_logging(ignore_errors=(CliError,)):                       │
│   255 │   │   async with meltano_stdout.redirect_stdout(), meltano_stderr.redirect_stderr():     │
│   256 │   │   │   try:                                                                           │
│ ❱ 257 │   │   │   │   yield                                                                      │
│   258 │   │   │   except CliError as err:                                                        │
│   259 │   │   │   │   err.print()                                                                │
│   260 │   │   │   │   raise                                                                      │
│                                                                                                  │
│ ╭─────────────────────────────────────────── locals ───────────────────────────────────────────╮ │
│ │            log = <BoundLogger(context={'name': 'meltano', 'run_id':                          │ │
│ │                  'db41504f-d251-4bd1-8f0e-ef4a44bef9e9', 'state_id':                         │ │
│ │                  '2022-12-02T183927--tap-workramp--target-bigquery'}, processors=[<function  │ │
│ │                  add_log_level at 0x103833700>,                                              │ │
│ │                  <structlog.stdlib.PositionalArgumentsFormatter object at 0x106b248e0>,      │ │
│ │                  <structlog.processors.TimeStamper object at 0x1044caf00>,                   │ │
│ │                  <structlog.processors.StackInfoRenderer object at 0x106b247c0>, <function   │ │
│ │                  format_exc_info at 0x1038454c0>, <function                                  │ │
│ │                  ProcessorFormatter.wrap_for_formatter at 0x103ec0c10>])>                    │ │
│ │ meltano_stderr = <meltano.core.logging.output_logger.Out object at 0x114357f70>              │ │
│ │ meltano_stdout = <meltano.core.logging.output_logger.Out object at 0x114357820>              │ │
│ │  output_logger = <meltano.core.logging.output_logger.OutputLogger object at 0x114346850>     │ │
│ ╰──────────────────────────────────────────────────────────────────────────────────────────────╯ │
│                                                                                                  │
│ /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_2/lib/python3.8/site-packages/m │
│ eltano/cli/elt.py:278 in _run_elt                                                                │
│                                                                                                  │
│   275 │   │   │   if elt_context.only_transform:                                                 │
│   276 │   │   │   │   log.info("Extract & load skipped.")                                        │
│   277 │   │   │   else:                                                                          │
│ ❱ 278 │   │   │   │   await _run_extract_load(log, elt_context, output_logger)                   │
│   279 │   │   │                                                                                  │
│   280 │   │   │   if elt_context.transformer:                                                    │
│   281 │   │   │   │   await _run_transform(log, elt_context, output_logger)                      │
│                                                                                                  │
│ ╭─────────────────────────────────────────── locals ───────────────────────────────────────────╮ │
│ │ context_builder = <meltano.core.elt_context.ELTContextBuilder object at 0x1142fdf70>         │ │
│ │     elt_context = <meltano.core.elt_context.ELTContext object at 0x114373a00>                │ │
│ │             log = <BoundLogger(context={'name': 'meltano', 'run_id':                         │ │
│ │                   'db41504f-d251-4bd1-8f0e-ef4a44bef9e9', 'state_id':                        │ │
│ │                   '2022-12-02T183927--tap-workramp--target-bigquery'}, processors=[<function │ │
│ │                   add_log_level at 0x103833700>,                                             │ │
│ │                   <structlog.stdlib.PositionalArgumentsFormatter object at 0x106b248e0>,     │ │
│ │                   <structlog.processors.TimeStamper object at 0x1044caf00>,                  │ │
│ │                   <structlog.processors.StackInfoRenderer object at 0x106b247c0>, <function  │ │
│ │                   format_exc_info at 0x1038454c0>, <function                                 │ │
│ │                   ProcessorFormatter.wrap_for_formatter at 0x103ec0c10>])>                   │ │
│ │   output_logger = <meltano.core.logging.output_logger.OutputLogger object at 0x114346850>    │ │
│ │         tracker = <meltano.core.tracking.tracker.Tracker object at 0x1031455b0>              │ │
│ ╰──────────────────────────────────────────────────────────────────────────────────────────────╯ │
│                                                                                                  │
│ /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_2/lib/python3.8/site-packages/m │
│ eltano/cli/elt.py:333 in _run_extract_load                                                       │
│                                                                                                  │
│   330 │   try:                                                                                   │
│   331 │   │   with extractor_log.line_writer() as extractor_log_writer, loader_log.line_writer   │
│   332 │   │   │   with extractor_out_writer_ctxmgr() as extractor_out_writer, loader_out_write   │
│ ❱ 333 │   │   │   │   await singer_runner.run(                                                   │
│   334 │   │   │   │   │   **kwargs,                                                              │
│   335 │   │   │   │   │   extractor_log=extractor_log_writer,                                    │
│   336 │   │   │   │   │   loader_log=loader_log_writer,                                          │
│                                                                                                  │
│ ╭─────────────────────────────────────────── locals ───────────────────────────────────────────╮ │
│ │                 elt_context = <meltano.core.elt_context.ELTContext object at 0x114373a00>    │ │
│ │                   extractor = 'tap-workramp'                                                 │ │
│ │               extractor_log = <meltano.core.logging.output_logger.Out object at 0x1143f58e0> │ │
│ │        extractor_log_writer = <meltano.core.logging.output_logger.LineWriter object at       │ │
│ │                               0x1143f58b0>                                                   │ │
│ │        extractor_out_writer = None                                                           │ │
│ │ extractor_out_writer_ctxmgr = <class 'contextlib.nullcontext'>                               │ │
│ │                      kwargs = {}                                                             │ │
│ │                      loader = 'target-bigquery'                                              │ │
│ │                  loader_log = <meltano.core.logging.output_logger.Out object at 0x1143f56d0> │ │
│ │           loader_log_writer = <meltano.core.logging.output_logger.LineWriter object at       │ │
│ │                               0x1143f5550>                                                   │ │
│ │           loader_out_writer = None                                                           │ │
│ │    loader_out_writer_ctxmgr = <class 'contextlib.nullcontext'>                               │ │
│ │                         log = <BoundLogger(context={'name': 'meltano', 'run_id':             │ │
│ │                               'db41504f-d251-4bd1-8f0e-ef4a44bef9e9', 'state_id':            │ │
│ │                               '2022-12-02T183927--tap-workramp--target-bigquery'},           │ │
│ │                               processors=[<function add_log_level at 0x103833700>,           │ │
│ │                               <structlog.stdlib.PositionalArgumentsFormatter object at       │ │
│ │                               0x106b248e0>, <structlog.processors.TimeStamper object at      │ │
│ │                               0x1044caf00>, <structlog.processors.StackInfoRenderer object   │ │
│ │                               at 0x106b247c0>, <function format_exc_info at 0x1038454c0>,    │ │
│ │                               <function ProcessorFormatter.wrap_for_formatter at             │ │
│ │                               0x103ec0c10>])>                                                │ │
│ │               output_logger = <meltano.core.logging.output_logger.OutputLogger object at     │ │
│ │                               0x114346850>                                                   │ │
│ │               singer_runner = <meltano.core.runner.singer.SingerRunner object at             │ │
│ │                               0x1143f52b0>                                                   │ │
│ │                  stderr_log = <BoundLogger(context={'run_id':                                │ │
│ │                               'db41504f-d251-4bd1-8f0e-ef4a44bef9e9', 'state_id':            │ │
│ │                               '2022-12-02T183927--tap-workramp--target-bigquery', 'stdio':   │ │
│ │                               'stderr'}, processors=[<function add_log_level at              │ │
│ │                               0x103833700>, <structlog.stdlib.PositionalArgumentsFormatter   │ │
│ │                               object at 0x106b248e0>, <structlog.processors.TimeStamper      │ │
│ │                               object at 0x1044caf00>,                                        │ │
│ │                               <structlog.processors.StackInfoRenderer object at              │ │
│ │                               0x106b247c0>, <function format_exc_info at 0x1038454c0>,       │ │
│ │                               <function ProcessorFormatter.wrap_for_formatter at             │ │
│ │                               0x103ec0c10>])>                                                │ │
│ ╰──────────────────────────────────────────────────────────────────────────────────────────────╯ │
│                                                                                                  │
│ /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_2/lib/python3.8/site-packages/m │
│ eltano/core/runner/singer.py:220 in run                                                          │
│                                                                                                  │
│   217 │   │                                                                                      │
│   218 │   │   async with tap.prepared(self.context.session):                                     │
│   219 │   │   │   async with target.prepared(self.context.session):                              │
│ ❱ 220 │   │   │   │   await self.invoke(                                                         │
│   221 │   │   │   │   │   tap,                                                                   │
│   222 │   │   │   │   │   target,                                                                │
│   223 │   │   │   │   │   extractor_log=extractor_log,                                           │
│                                                                                                  │
│ ╭─────────────────────────────────────── locals ────────────────────────────────────────╮        │
│ │ extractor_log = <meltano.core.logging.output_logger.LineWriter object at 0x1143f58b0> │        │
│ │ extractor_out = None                                                                  │        │
│ │    loader_log = <meltano.core.logging.output_logger.LineWriter object at 0x1143f5550> │        │
│ │    loader_out = None                                                                  │        │
│ │          self = <meltano.core.runner.singer.SingerRunner object at 0x1143f52b0>       │        │
│ │           tap = <meltano.core.plugin_invoker.PluginInvoker object at 0x1143f5fd0>     │        │
│ │        target = <meltano.core.plugin_invoker.PluginInvoker object at 0x1143f5ca0>     │        │
│ ╰───────────────────────────────────────────────────────────────────────────────────────╯        │
│                                                                                                  │
│ /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_2/lib/python3.8/site-packages/m │
│ eltano/core/runner/singer.py:148 in invoke                                                       │
│                                                                                                  │
│   145 │   │   │   │   │   )                                                                      │
│   146 │   │   │   │                                                                              │
│   147 │   │   │   │   failed_future = output_futures_failed.pop()                                │
│ ❱ 148 │   │   │   │   raise failed_future.exception()                                            │
│   149 │   │   │   else:                                                                          │
│   150 │   │   │   │   # If all of the output handlers completed without raising an exception,    │
│   151 │   │   │   │   # we still need to wait for the tap or target to complete.                 │
│                                                                                                  │
│ ╭─────────────────────────────────────────── locals ───────────────────────────────────────────╮ │
│ │                       _ = {                                                                  │ │
│ │                           │   <Task pending name='Task-12' coro=<capture_subprocess_output() │ │
│ │                           running at                                                         │ │
│ │                           /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_… │ │
│ │                           wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at      │ │
│ │                           0x114373370>()]>>                                                  │ │
│ │                           }                                                                  │ │
│ │                    done = {                                                                  │ │
│ │                           │   <Task finished name='Task-17' coro=<wait() done, defined at    │ │
│ │                           /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/a… │ │
│ │                           result=({<Task finishe...> result=None>, <Task finishe...>         │ │
│ │                           result=None>, <Task finishe...Broken pipe')>}, {<Task              │ │
│ │                           pending...14373370>()]>>})>,                                       │ │
│ │                           │   <Task finished name='Task-16' coro=<Process.wait() done,       │ │
│ │                           defined at                                                         │ │
│ │                           /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/a… │ │
│ │                           result=2>                                                          │ │
│ │                           }                                                                  │ │
│ │           extractor_log = <meltano.core.logging.output_logger.LineWriter object at           │ │
│ │                           0x1143f58b0>                                                       │ │
│ │           extractor_out = None                                                               │ │
│ │           failed_future = <Task finished name='Task-11' coro=<capture_subprocess_output()    │ │
│ │                           done, defined at                                                   │ │
│ │                           /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_… │ │
│ │                           exception=BrokenPipeError(32, 'Broken pipe')>                      │ │
│ │       line_length_limit = 5242880                                                            │ │
│ │              loader_log = <meltano.core.logging.output_logger.LineWriter object at           │ │
│ │                           0x1143f5550>                                                       │ │
│ │              loader_out = None                                                               │ │
│ │ output_exception_future = <Task finished name='Task-17' coro=<wait() done, defined at        │ │
│ │                           /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/a… │ │
│ │                           result=({<Task finishe...> result=None>, <Task finishe...>         │ │
│ │                           result=None>, <Task finishe...Broken pipe')>}, {<Task              │ │
│ │                           pending...14373370>()]>>})>                                        │ │
│ │     output_futures_done = {                                                                  │ │
│ │                           │   <Task finished name='Task-13'                                  │ │
│ │                           coro=<capture_subprocess_output() done, defined at                 │ │
│ │                           /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_… │ │
│ │                           result=None>,                                                      │ │
│ │                           │   <Task finished name='Task-14'                                  │ │
│ │                           coro=<capture_subprocess_output() done, defined at                 │ │
│ │                           /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_… │ │
│ │                           result=None>,                                                      │ │
│ │                           │   <Task finished name='Task-11'                                  │ │
│ │                           coro=<capture_subprocess_output() done, defined at                 │ │
│ │                           /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_… │ │
│ │                           exception=BrokenPipeError(32, 'Broken pipe')>                      │ │
│ │                           }                                                                  │ │
│ │   output_futures_failed = []                                                                 │ │
│ │                   p_tap = <Process 81252>                                                    │ │
│ │                p_target = <Process 81253>                                                    │ │
│ │                    self = <meltano.core.runner.singer.SingerRunner object at 0x1143f52b0>    │ │
│ │      stream_buffer_size = 10485760                                                           │ │
│ │                     tap = <meltano.core.plugin_invoker.PluginInvoker object at 0x1143f5fd0>  │ │
│ │             tap_outputs = [                                                                  │ │
│ │                           │   <StreamWriter transport=<_UnixWritePipeTransport closed fd=18  │ │
│ │                           closed>>                                                           │ │
│ │                           ]                                                                  │ │
│ │      tap_process_future = <Task pending name='Task-15' coro=<Process.wait() running at       │ │
│ │                           /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/a… │ │
│ │                           wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at      │ │
│ │                           0x11440b4c0>()]>>                                                  │ │
│ │       tap_stderr_future = <Task pending name='Task-12' coro=<capture_subprocess_output()     │ │
│ │                           running at                                                         │ │
│ │                           /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_… │ │
│ │                           wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at      │ │
│ │                           0x114373370>()]>>                                                  │ │
│ │       tap_stdout_future = <Task finished name='Task-11' coro=<capture_subprocess_output()    │ │
│ │                           done, defined at                                                   │ │
│ │                           /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_… │ │
│ │                           exception=BrokenPipeError(32, 'Broken pipe')>                      │ │
│ │                  target = <meltano.core.plugin_invoker.PluginInvoker object at 0x1143f5ca0>  │ │
│ │          target_outputs = [                                                                  │ │
│ │                           │   <meltano.core.plugin.singer.target.BookmarkWriter object at    │ │
│ │                           0x1143f51f0>                                                       │ │
│ │                           ]                                                                  │ │
│ │   target_process_future = <Task finished name='Task-16' coro=<Process.wait() done, defined   │ │
│ │                           at                                                                 │ │
│ │                           /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/a… │ │
│ │                           result=2>                                                          │ │
│ │    target_stderr_future = <Task finished name='Task-14' coro=<capture_subprocess_output()    │ │
│ │                           done, defined at                                                   │ │
│ │                           /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_… │ │
│ │                           result=None>                                                       │ │
│ │    target_stdout_future = <Task finished name='Task-13' coro=<capture_subprocess_output()    │ │
│ │                           done, defined at                                                   │ │
│ │                           /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_… │ │
│ │                           result=None>                                                       │ │
│ ╰──────────────────────────────────────────────────────────────────────────────────────────────╯ │
│                                                                                                  │
│ /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_2/lib/python3.8/site-packages/m │
│ eltano/core/logging/utils.py:240 in capture_subprocess_output                                    │
│                                                                                                  │
│   237 │   │   │   continue                                                                       │
│   238 │   │                                                                                      │
│   239 │   │   for writer in line_writers:                                                        │
│ ❱ 240 │   │   │   if not await _write_line_writer(writer, line):                                 │
│   241 │   │   │   │   # If the destination stream is closed, we can stop capturing output.       │
│   242 │   │   │   │   return                                                                     │
│   243                                                                                            │
│                                                                                                  │
│ ╭─────────────────────────────────────────── locals ───────────────────────────────────────────╮ │
│ │         line = b'{"type": "RECORD", "stream": "path_assignments", "record": {"id":           │ │
│ │                "d509845c-f93b'+338                                                           │ │
│ │ line_writers = (<StreamWriter transport=<_UnixWritePipeTransport closed fd=18 closed>>,)     │ │
│ │       reader = <StreamReader 693222 bytes limit=5242880 transport=<_UnixReadPipeTransport    │ │
│ │                fd=9 polling>>                                                                │ │
│ │       writer = <StreamWriter transport=<_UnixWritePipeTransport closed fd=18 closed>>        │ │
│ ╰──────────────────────────────────────────────────────────────────────────────────────────────╯ │
│                                                                                                  │
│ /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_2/lib/python3.8/site-packages/m │
│ eltano/core/logging/utils.py:209 in _write_line_writer                                           │
│                                                                                                  │
│   206 │   │   │   await writer.drain()                                                           │
│   207 │   │   except (BrokenPipeError, ConnectionResetError):                                    │
│   208 │   │   │   with suppress(AttributeError):  # `wait_closed` is Python 3.7+                 │
│ ❱ 209 │   │   │   │   await writer.wait_closed()                                                 │
│   210 │   │   │                                                                                  │
│   211 │   │   │   return False                                                                   │
│   212 │   else:                                                                                  │
│                                                                                                  │
│ ╭─────────────────────────────────────────── locals ───────────────────────────────────────────╮ │
│ │   line = [**redacted by AJ**]                                                                 │ │
│ │ writer = <StreamWriter transport=<_UnixWritePipeTransport closed fd=18 closed>>              │ │
│ ╰──────────────────────────────────────────────────────────────────────────────────────────────╯ │
│                                                                                                  │
│ /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/streams.py:359 in        │
│ wait_closed                                                                                      │
│                                                                                                  │
│   356 │   │   return self._transport.is_closing()                                                │
│   357 │                                                                                          │
│   358 │   async def wait_closed(self):                                                           │
│ ❱ 359 │   │   await self._protocol._get_close_waiter(self)                                       │
│   360 │                                                                                          │
│   361 │   def get_extra_info(self, name, default=None):                                          │
│   362 │   │   return self._transport.get_extra_info(name, default)                               │
│                                                                                                  │
│ ╭─────────────────────────────────── locals ────────────────────────────────────╮                │
│ │ self = <StreamWriter transport=<_UnixWritePipeTransport closed fd=18 closed>> │                │
│ ╰───────────────────────────────────────────────────────────────────────────────╯                │
│                                                                                                  │
│ /Users/stephen.hazeltine/Documents/Projects/meltano/meltano_venv_2/lib/python3.8/site-packages/m │
│ eltano/core/logging/utils.py:206 in _write_line_writer                                           │
│                                                                                                  │
│   203 │   if isinstance(writer, asyncio.StreamWriter):                                           │
│   204 │   │   try:  # noqa: WPS229                                                               │
│   205 │   │   │   writer.write(line)                                                             │
│ ❱ 206 │   │   │   await writer.drain()                                                           │
│   207 │   │   except (BrokenPipeError, ConnectionResetError):                                    │
│   208 │   │   │   with suppress(AttributeError):  # `wait_closed` is Python 3.7+                 │
│   209 │   │   │   │   await writer.wait_closed()                                                 │
│                                                                                                  │
│ ╭─────────────────────────────────────────── locals ───────────────────────────────────────────╮ │
│ │   line = [**redacted by AJ**]                                                                 │ │
│ │ writer = <StreamWriter transport=<_UnixWritePipeTransport closed fd=18 closed>>              │ │
│ ╰──────────────────────────────────────────────────────────────────────────────────────────────╯ │
│                                                                                                  │
│ /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/streams.py:387 in drain  │
│                                                                                                  │
│   384 │   │   │   # in a loop would never call connection_lost(), so it                          │
│   385 │   │   │   # would not see an error when the socket is closed.                            │
│   386 │   │   │   await sleep(0)                                                                 │
│ ❱ 387 │   │   await self._protocol._drain_helper()                                               │
│   388                                                                                            │
│   389                                                                                            │
│   390 class StreamReader:                                                                        │
│                                                                                                  │
│ ╭─────────────────────────────────── locals ────────────────────────────────────╮                │
│ │ self = <StreamWriter transport=<_UnixWritePipeTransport closed fd=18 closed>> │                │
│ ╰───────────────────────────────────────────────────────────────────────────────╯                │
│                                                                                                  │
│ /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/streams.py:197 in        │
│ _drain_helper                                                                                    │
│                                                                                                  │
│   194 │   │   assert waiter is None or waiter.cancelled()                                        │
│   195 │   │   waiter = self._loop.create_future()                                                │
│   196 │   │   self._drain_waiter = waiter                                                        │
│ ❱ 197 │   │   await waiter                                                                       │
│   198 │                                                                                          │
│   199 │   def _get_close_waiter(self, stream):                                                   │
│   200 │   │   raise NotImplementedError                                                          │
│                                                                                                  │
│ ╭─────────────────────────────────────────── locals ───────────────────────────────────────────╮ │
│ │   self = <SubprocessStreamProtocol stdin=<StreamWriter transport=<_UnixWritePipeTransport    │ │
│ │          closed fd=18 closed>> stdout=<StreamReader eof limit=5242880                        │ │
│ │          transport=<_UnixReadPipeTransport closed fd=19 closed>> stderr=<StreamReader eof    │ │
│ │          limit=5242880 transport=<_UnixReadPipeTransport closed fd=21 closed>>>              │ │
│ │ waiter = <Future finished exception=BrokenPipeError(32, 'Broken pipe')>                      │ │
│ ╰──────────────────────────────────────────────────────────────────────────────────────────────╯ │
│                                                                                                  │
│ /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/unix_events.py:687 in    │
│ _write_ready                                                                                     │
│                                                                                                  │
│    684 │   │   assert self._buffer, 'Data should not be empty'                                   │
│    685 │   │                                                                                     │
│    686 │   │   try:                                                                              │
│ ❱  687 │   │   │   n = os.write(self._fileno, self._buffer)                                      │
│    688 │   │   except (BlockingIOError, InterruptedError):                                       │
│    689 │   │   │   pass                                                                          │
│    690 │   │   except (SystemExit, KeyboardInterrupt):                                           │
│                                                                                                  │
│ ╭─────────────────────── locals ───────────────────────╮                                         │
│ │ self = <_UnixWritePipeTransport closed fd=18 closed> │                                         │
│ ╰──────────────────────────────────────────────────────╯                                         │
╰──────────────────────────────────────────────────────────────────────────────────────────────────╯
BrokenPipeError: [Errno 32] Broken pipe

Related discussions
@tayloramurphy tayloramurphy added kind/Bug Something isn't working valuestream/Meltano labels Dec 2, 2022
@aaronsteers aaronsteers mentioned this issue Feb 17, 2023
Open
@aaronsteers aaronsteers mentioned this issue Feb 17, 2023
Open
@stale
Copy link
stale bot commented Jun 17, 2023

This has been marked as stale because it is unassigned, and has not had recent activity. It will be closed after 21 days if no further activity occurs. If this should never go stale, please add the evergreen label, or request that it be added.

@stale stale bot added the stale label Jun 17, 2023
@edgarrmondragon
Copy link
Collaborator

Still relevant

@stale stale bot removed the stale label Jun 19, 2023
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees
No one assigned
Labels
kind/Bug Something isn't working urgency/High valuestream/Meltano
Projects
None yet
Milestone
No milestone
Development

No branches or pull requests
3 participants
@tayloramurphy @edgarrmondragon @aaronsteers