redpanda-data / redpanda-data/connect
ERROR logs on pipeline shutdown when using Python Redpanda Connect plugin
Nobody has claimed this yet.
- Dominant language
- Go
- Stars
- 8.8k
- Forks
- 969
- Avg merge
- 1d 13h
- Merged PRs (30d)
- 64
Description
Hi! I’m experimenting with the new Redpanda Connect Plugins framework. During pipeline shutdown, I see the following ERROR logs:
INFO[2026-01-23T07:44:07Z] Running main config from specified file @service=redpanda-connect benthos_version=4.78.0 path=connect.yaml
INFO[2026-01-23T07:44:07Z] Successfully loaded Redpanda license @service=redpanda-connect expires_at="2036-01-21T07:44:07Z" license_org="" license_type=open_source
INFO[2026-01-23T07:44:07Z] Listening for HTTP requests at: http://0.0.0.0:4195 @service=redpanda-connect
INFO[2026-01-23T07:44:08Z] Launching a Redpanda Connect instance, use CTRL+C to close @service=redpanda-connect
INFO[2026-01-23T07:44:08Z] Output type file is now active @service=redpanda-connect label="" path=root.output
INFO[2026-01-23T07:44:08Z] Input type generate is now active @service=redpanda-connect label="" path=root.input
INFO[2026-01-23T07:44:09Z] ERROR:grpc._cython.cygrpc:Exception not handled by _handle_exceptions in servicer method [/redpanda.runtime.v1alpha1.BatchProcessorService/Close] @service=redpanda-connect label="" path=root.pipeline.processors.0 source=stderr
INFO[2026-01-23T07:44:09Z] Traceback (most recent call last): @service=redpanda-connect label="" path=root.pipeline.processors.0 source=stderr
INFO[2026-01-23T07:44:09Z] File "src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi", line 787, in grpc._cython.cygrpc._schedule_rpc_coro @service=redpanda-connect label="" path=root.pipeline.processors.0 source=stderr
INFO[2026-01-23T07:44:09Z] asyncio.exceptions.CancelledError @service=redpanda-connect label="" path=root.pipeline.processors.0 source=stderr
INFO[2026-01-23T07:44:09Z] Traceback (most recent call last): @service=redpanda-connect label="" path=root.pipeline.processors.0 source=stderr
INFO[2026-01-23T07:44:09Z] File "src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi", line 787, in grpc._cython.cygrpc._schedule_rpc_coro @service=redpanda-connect label="" path=root.pipeline.processors.0 source=stderr
INFO[2026-01-23T07:44:09Z] asyncio.exceptions.CancelledError @service=redpanda-connect label="" path=root.pipeline.processors.0 source=stderr
INFO[2026-01-23T07:44:09Z] Successfully loaded Redpanda Connect RPC plugin @service=redpanda-connect label="" path=root.pipeline.processors.0 source=stdout
INFO[2026-01-23T07:44:09Z] Pipeline has terminated. Shutting down the service @service=redpanda-connect
The pipeline itself works as expected. Is this expected behavior, and is there a way to avoid these ERROR logs during shutdown?
Here’s the setup:
connect.yaml
input:
generate:
mapping: |
root = {
"message": fake("sentence")
}
interval: ""
count: 1000
pipeline:
processors:
- yell: {}
output:
file:
codec: lines
path: ./processed/output.jsonl
plugin.yaml
name: yell
summary: yells back what you give it
command: ["uv", "run", "--no-sync", "--frozen", "main.py"]
type: processor
fields: []
main.py
import asyncio
import logging
from datetime import datetime, timezone
from redpanda_connect import Message, processor, processor_main
logging.getLogger("grpc").setLevel(logging.ERROR)
logging.getLogger("grpc._cython.cygrpc").setLevel(logging.ERROR)
logging.getLogger("redpanda_connect._grpc").setLevel(logging.ERROR)
logger = logging.getLogger(__name__)
@processor
def yell(msg: Message) -> Message:
try:
data = msg.payload
data["message"] = data["message"].upper()
data["yelled_at"] = datetime.now(timezone.utc)
return msg
except Exception as err:
logger.error(f"Error processing message: {err}", exc_info=True)
return msg
if __name__ == "__main__":
asyncio.run(processor_main(yell))
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Reproduce the shutdown behavior with connect.yaml, plugin.yaml, and main.py, using the provided pipeline and Python processor. Start by tracing processor_main(yell) and the gRPC BatchProcessorService Close path; done means determining whether the CancelledError logs are expected and, if not, preventing them during normal shutdown.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- grpc, python
- Domain
- stream-processing
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100