redpanda (kafka (eventhubs)) input stalling for exactly 15 minutes

[JoranDox]

Hi, we're seeing occasional input stalls in our redpanda connect pods consuming eventhubs data. Crucially: logstash (which is consuming the same data in parallel while we're trying to replace it with redpanda connect) is not seeing this, so it can't be purely an eventhubs problem.

Relevant parts of our setup:

• right before the data enters eventhubs, we timestamp it
• at the moment we start processing it in logstash/connect we timestamp it again and compute the diff with the first timestamp to monitor processinglag
• we also tag the data with the eventhubs partition
• both output to the same elasticsearch in different indices, and we monitor processinglag, and connect is seeing peaks of 900s on specific partitions

enabling debug logging (see collapsible below for detailed log) shows about this:

• happily processing, printing something with every batch sent
• something happens like a partition rebalance, takes at most some seconds to finish
• 15 minutes of nothing
• happy processing again

what we've tried:

• we used to use the kafka input at version 1.0.0 (probably an old default we copied since forever), so we switched this to the redpanda input instead, same thing happens
• changed settings from defaults (some of these variables play/interfere with each other so have changed multiple times) but no improvement observed
  • session_timeout: default 1m → 10s
  • conn_idle_timeout: raised to 20s
  • request_timeout_overhead: raised to 40s
  • rebalance_timeout: pinned explicitly at its default 45s (not actually changed, just made explicit)
  • tcp.tcp_user_timeout: added (default 0/disabled) → 8s
  • tcp.keep_alive.idle: raised default 15s → 30s
  • unordered_processing on & off again
  • max_yield_batch_bytes: 10000000
  • fetch_max_partition_bytes: 10000000
  • partition_buffer_bytes: 12000000

again, crucially the logstash is not experiencing these stalls so it's almost certain it's connect-specific, but it's hard to debug especially since the debuglogs are showing nothing at all in that 15 minute window

Any suggestions in helping debugging or potentially relevant 15-minute timeouts (or 5-minute timeouts with 2 retries etc etc) are very welcome!

debug logs

time="2026-07-22T01:50:51Z" level=debug msg="Successfully dispatched [3] documents in 45ms (66.666667 docs/sec)" @service=benthos label="" path=root.output.broker.outputs.0
time="2026-07-22T01:50:51Z" level=debug msg="Successfully dispatched [13] documents in 123ms (105.691057 docs/sec)" @service=benthos label="" path=root.output.broker.outputs.0
time="2026-07-22T01:50:55Z" level=debug msg="Successfully dispatched [7] documents in 12ms (583.333333 docs/sec)" @service=benthos label="" path=root.output.broker.outputs.0
time="2026-07-22T01:51:01Z" level=warning msg="Kafka broker read failed" @service=benthos addr="prodmheventhubs.servicebus.windows.net:9093" broker=0 err="context deadline exceeded" label=kafka_input path
=root.input.broker.inputs.0 request=DescribeGroups
time="2026-07-22T01:51:01Z" level=debug msg="Failed to fetch group lags: request DescribeGroups has 1 separate shard errors, first: context deadline exceeded" @service=benthos label=kafka_input path=root.
input.broker.inputs.0
time="2026-07-22T01:51:01Z" level=error msg="default commit failed" @service=benthos err="the broker connection has died and the request will be retried on a new connection" group=traffic-streams-prod-mhb
e-events-consumer-v3 label=kafka_input path=root.input.broker.inputs.0
time="2026-07-22T01:51:05Z" level=debug msg="heartbeat errored" @service=benthos err="REBALANCE_IN_PROGRESS: The group is rebalancing, so a rejoin is needed." group=traffic-streams-prod-mhbe-events-consum
er-v3 label=kafka_input path=root.input.broker.inputs.0
time="2026-07-22T01:51:05Z" level=debug msg="consumer calling onRevoke at the end of a session; consumer did not change any client-side subscription" @service=benthos group=traffic-streams-prod-mhbe-event
s-consumer-v3 label=kafka_input path=root.input.broker.inputs.0
time="2026-07-22T01:51:05Z" level=debug msg="joining group" @service=benthos group=traffic-streams-prod-mhbe-events-consumer-v3 label=kafka_input path=root.input.broker.inputs.0
time="2026-07-22T01:51:06Z" level=warning msg="Kafka broker read failed" @service=benthos addr="prodmheventhubs.servicebus.windows.net:9093" broker=-2147483648 err="context deadline exceeded" label=kafka_input path=root.input.broker.inputs.0 request=Metadata
time="2026-07-22T01:51:06Z" level=debug msg="Failed to fetch group lags: context deadline exceeded" @service=benthos label=kafka_input path=root.input.broker.inputs.0
time="2026-07-22T01:51:08Z" level=debug msg=joined @service=benthos generation=74967704 group=traffic-streams-prod-mhbe-events-consumer-v3 instance_id="<nil>" label=kafka_input leader=false member_id="prodmheventhubs.servicebus.windows.net:c:traffic-streams-prod-mhbe-events-consumer-v3:I:redpanda-connect-19aab9f93a5d4b9facdd45c8ffcd0651" path=root.input.broker.inputs.0
time="2026-07-22T01:51:08Z" level=debug msg=syncing @service=benthos group=traffic-streams-prod-mhbe-events-consumer-v3 label=kafka_input path=root.input.broker.inputs.0 protocol=cooperative-sticky protocol_type=consumer
time="2026-07-22T01:51:08Z" level=debug msg=synced @service=benthos assigned="processedevent[1]" group=traffic-streams-prod-mhbe-events-consumer-v3 label=kafka_input path=root.input.broker.inputs.0
time="2026-07-22T01:51:08Z" level=debug msg="new group session begun" @service=benthos added="" group=traffic-streams-prod-mhbe-events-consumer-v3 label=kafka_input lost="" path=root.input.broker.inputs.0
time="2026-07-22T01:51:08Z" level=debug msg="beginning heartbeat loop" @service=benthos group=traffic-streams-prod-mhbe-events-consumer-v3 label=kafka_input path=root.input.broker.inputs.0
time="2026-07-22T01:51:10Z" level=debug msg="metadata update triggered" @service=benthos label=kafka_input path=root.input.broker.inputs.0 why="opportunistic load during source backoff: read tcp 10.244.189.38:39672->20.86.89.184:9093: i/o timeout"
time="2026-07-22T02:06:48Z" level=debug msg="Successfully dispatched [79] documents in 72ms (1097.222222 docs/sec)" @service=benthos label="" path=root.output.broker.outputs.0
time="2026-07-22T02:06:55Z" level=debug msg="Successfully dispatched [2288] documents in 4.022s (568.871208 docs/sec)" @service=benthos label="" path=root.output.broker.outputs.0
time="2026-07-22T02:06:57Z" level=debug msg="Successfully dispatched [546] documents in 1.267s (430.939227 docs/sec)" @service=benthos label="" path=root.output.broker.outputs.0
time="2026-07-22T02:06:59Z" level=debug msg="Successfully dispatched [804] documents in 1.367s (588.149232 docs/sec)" @service=benthos label="" path=root.output.broker.outputs.0
time="2026-07-22T02:07:05Z" level=debug msg="Successfully dispatched [18] documents in 25ms (720.000000 docs/sec)" @service=benthos label="" path=root.output.broker.outputs.0
time="2026-07-22T02:07:10Z" level=debug msg="Successfully dispatched [7] documents in 12ms (583.333333 docs/sec)" @service=benthos label="" path=root.output.broker.outputs.0
time="2026-07-22T02:07:11Z" level=debug msg="Successfully dispatched [6] documents in 18ms (333.333333 docs/sec)" @service=benthos label="" path=root.output.broker.outputs.0

[JoranDox]

forgot to mention: we added a workaround adding a metric with the timestamp of the last time each partition got new data, and if this goes too stale, we just kill the pod, but that's not ideal, and if nothing else I'd like to be able to force a reconnect from within redpanda connect perhaps, which hopefully could be faster than k8s doing a complete restart?

[JoranDox]

EDIT: ignore, see comment below

I got a result from where I think it's happening (40+s stall at this point, while we expect to see over 1 message/s): call to http://localhost:4195/debug/pprof/goroutine?debug=2 , let me know if this is useful

Details

goroutine 5041 [running]:
runtime/pprof.writeGoroutineStacks({0xbaad240, 0x22116830cf00})
    /opt/hostedtoolcache/go/1.26.5/x64/src/runtime/pprof/pprof.go:819 +0x6b
runtime/pprof.writeGoroutine({0xbaad240?, 0x22116830cf00?}, 0x2211674097d8?)
    /opt/hostedtoolcache/go/1.26.5/x64/src/runtime/pprof/pprof.go:782 +0x25
runtime/pprof.(*Profile).WriteTo(0x1429c270?, {0xbaad240?, 0x22116830cf00?}, 0xc?)
    /opt/hostedtoolcache/go/1.26.5/x64/src/runtime/pprof/pprof.go:408 +0x149
net/http/pprof.handler.ServeHTTP({0xb54b8ab, 0x9}, {0xbb4ed60, 0x22116830cf00}, 0x221166013b80)
    /opt/hostedtoolcache/go/1.26.5/x64/src/net/http/pprof/pprof.go:273 +0x52a
github.com/redpanda-data/benthos/v4/internal/api.(*Type).RegisterEndpoint.func1({0xbb4ed60, 0x22116830cf00}, 0x221166013b80)
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/api/api.go:257 +0xce
net/http.HandlerFunc.ServeHTTP(0x221166013400?, {0xbb4ed60?, 0x22116830cf00?}, 0x22116675d1aa?)
    /opt/hostedtoolcache/go/1.26.5/x64/src/net/http/server.go:2286 +0x29
github.com/gorilla/mux.(*Router).ServeHTTP(0x221166f8f440, {0xbb4ed60, 0x22116830cf00}, 0x221166013180)
    /home/runner/go/pkg/mod/github.com/gorilla/mux@v1.8.1/mux.go:212 +0x1e2
net/http.serverHandler.ServeHTTP({0x221167914f40?}, {0xbb4ed60?, 0x22116830cf00?}, 0x1?)
    /opt/hostedtoolcache/go/1.26.5/x64/src/net/http/server.go:3311 +0x8e
net/http.(*conn).serve(0x221166d81950, {0xbb6a530, 0x221166d7c300})
    /opt/hostedtoolcache/go/1.26.5/x64/src/net/http/server.go:2073 +0x650
created by net/http.(*Server).Serve in goroutine 14
    /opt/hostedtoolcache/go/1.26.5/x64/src/net/http/server.go:3464 +0x485

goroutine 1 [select, 5 minutes]:
github.com/redpanda-data/benthos/v4/internal/cli/common.RunManagerUntilStopped(_, _, {{{0x221167ae0f80, 0xc}, 0x1, {0x221167ae0fb8, 0x8}, 0x1, {0x0, 0x0}, ...}, ...}, ...)
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/cli/common/manager.go:209 +0x409
github.com/redpanda-data/benthos/v4/internal/cli/common.RunService(0x221167355c00, 0x221165a92600, 0x0)
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/cli/common/service.go:110 +0xb51
github.com/redpanda-data/benthos/v4/internal/cli.App.func3(0x221167355c00)
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/cli/app.go:189 +0x18a
github.com/urfave/cli/v2.(*Command).Run(0x2211668f9b80, 0x221167355c00, {0x2211656f5920, 0x3, 0x3})
    /home/runner/go/pkg/mod/github.com/urfave/cli/v2@v2.27.7/command.go:276 +0x7c2
github.com/urfave/cli/v2.(*App).RunContext(0x221166551200, {0xbb6a070, 0x146538e0}, {0x2211656f5920, 0x3, 0x3})
    /home/runner/go/pkg/mod/github.com/urfave/cli/v2@v2.27.7/app.go:333 +0x5a5
github.com/redpanda-data/benthos/v4/public/service.RunCLIToCode({0xbb6a070, 0x146538e0}, {0x2211663ef5f0, 0x11, 0x9e49180?})
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/public/service/service.go:79 +0x165
github.com/redpanda-data/connect/v4/internal/cli.InitEnterpriseCLI({0xbaa1410, 0x10}, {0xba67160, 0x7}, {0xbaa1470, 0x14}, 0x22116734b700, {0x0, 0x0, 0x0})
    /home/runner/work/connect/connect/internal/cli/enterprise.go:245 +0x1cb9
main.main()
    /home/runner/work/connect/connect/cmd/redpanda-connect/main.go:34 +0x72

goroutine 8 [select]:
go.opencensus.io/stats/view.(*worker).start(0x221165682980)
    /home/runner/go/pkg/mod/go.opencensus.io@v0.24.0/stats/view/worker.go:292 +0xa5
created by go.opencensus.io/stats/view.init.0 in goroutine 1
    /home/runner/go/pkg/mod/go.opencensus.io@v0.24.0/stats/view/worker.go:34 +0x8d

goroutine 19 [select, 2 minutes]:
github.com/redpanda-data/benthos/v4/internal/pipeline.(*Pool).loop(0x221166b14360)
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/pipeline/pool.go:126 +0x37b
created by github.com/redpanda-data/benthos/v4/internal/pipeline.(*Pool).Consume in goroutine 1
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/pipeline/pool.go:150 +0x78

goroutine 14 [IO wait]:
internal/poll.runtime_pollWait(0x7f2cf956f200, 0x72)
    /opt/hostedtoolcache/go/1.26.5/x64/src/runtime/netpoll.go:351 +0x85
internal/poll.(*pollDesc).wait(0x22116613a000?, 0x900000036?, 0x0)
    /opt/hostedtoolcache/go/1.26.5/x64/src/internal/poll/fd_poll_runtime.go:84 +0x27
internal/poll.(*pollDesc).waitRead(...)
    /opt/hostedtoolcache/go/1.26.5/x64/src/internal/poll/fd_poll_runtime.go:89
internal/poll.(*FD).Accept(0x22116613a000)
    /opt/hostedtoolcache/go/1.26.5/x64/src/internal/poll/fd_unix.go:613 +0x28c
net.(*netFD).accept(0x22116613a000)
    /opt/hostedtoolcache/go/1.26.5/x64/src/net/fd_unix.go:150 +0x29
net.(*TCPListener).accept(0x2211662b8000)
    /opt/hostedtoolcache/go/1.26.5/x64/src/net/tcpsock_posix.go:159 +0x1b
net.(*TCPListener).Accept(0x2211662b8000)
    /opt/hostedtoolcache/go/1.26.5/x64/src/net/tcpsock.go:387 +0x30
net/http.(*Server).Serve(0x2211678f8000, {0xbb4f470, 0x2211662b8000})
    /opt/hostedtoolcache/go/1.26.5/x64/src/net/http/server.go:3434 +0x30c
net/http.(*Server).ListenAndServe(0x2211678f8000)
    /opt/hostedtoolcache/go/1.26.5/x64/src/net/http/server.go:3360 +0x72
github.com/redpanda-data/benthos/v4/internal/api.(*Type).ListenAndServe(0x221166414780)
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/api/api.go:282 +0xe7
github.com/redpanda-data/benthos/v4/internal/cli/common.newStoppableManager.func1()
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/cli/common/manager.go:239 +0x2d
created by github.com/redpanda-data/benthos/v4/internal/cli/common.newStoppableManager in goroutine 1
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/cli/common/manager.go:238 +0xe5

goroutine 15 [select, 2 minutes]:
github.com/redpanda-data/common-go/telemetry.(*Reporter).Run(0x221166d51680, {0xbb6a070, 0x146538e0})
    /home/runner/go/pkg/mod/github.com/redpanda-data/common-go/telemetry@v0.1.0/reporter.go:94 +0x368
github.com/redpanda-data/connect/v4/internal/telemetry.ActivateExporter.func2()
    /home/runner/work/connect/connect/internal/telemetry/telemetry.go:115 +0x25
created by github.com/redpanda-data/connect/v4/internal/telemetry.ActivateExporter in goroutine 1
    /home/runner/work/connect/connect/internal/telemetry/telemetry.go:115 +0x848

goroutine 157 [select]:
github.com/twmb/franz-go/pkg/kgo.(*groupConsumer).loopCommit(0x22116626a3c0)
    /home/runner/go/pkg/mod/github.com/twmb/franz-go@v1.20.7/pkg/kgo/consumer_group.go:2188 +0xd9
created by github.com/twmb/franz-go/pkg/kgo.(*groupConsumer).manage in goroutine 155
    /home/runner/go/pkg/mod/github.com/twmb/franz-go@v1.20.7/pkg/kgo/consumer_group.go:489 +0x226

goroutine 20 [select, 5 minutes]:
github.com/redpanda-data/benthos/v4/internal/impl/pure.(*indefiniteRetry).WaitForClose(0x8992b2264ac8992b?, {0xbb6a070, 0x146538e0})
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/impl/pure/output_retry.go:272 +0x70
github.com/redpanda-data/benthos/v4/internal/stream.(*Type).init.func1({0xbba5708?, 0x2211673fec80?})
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/stream/type.go:195 +0x3c
created by github.com/redpanda-data/benthos/v4/internal/stream.(*Type).init in goroutine 1
    /home/runner/go/pkg/mod/github.com/redpanda-data/benthos/v4@v4.77.0/internal/stream/type.go:192 +0x57a

goroutine 21 [sync.Cond.Wait, 2 minutes]:
sync.runtime_notifyListWait(0x2211663efbe0, 0x32)
    /opt/hostedtoolcache/go/1.26.5/x64/src/runtime/sema.go:617 +0x1b3
sync.(*Cond).Wait(0xbb6a568?)
    /opt/hostedtoolcache/go/1.26.5/x64/src/sync/cond.go:71 +0x73
github.com/redpanda-data/benthos/v4/internal/autoretry.(*List[...]).Shift(0xbc37ae0, {0xbb6a568, 0x221167

[JoranDox]

EDIT: ignore, see comment below

apologies, seems like the logs got truncated, here's two full logs I believe:

Logs-2026-07-31 15_04_02.txt

[JoranDox]

taking those logs back, that appears to actually be a light partition + not-round-robin partitioning in this case, I'm trying to see if I can replicate the 15 min stalls on one of our busier topics instead

[JoranDox]

update:

• couldn't replicate locally running only the franz kafka against our eventhubs, so back to the drawing board
• got some new logs from a 600s stall (after which we kill the pod) which imply it may be on the other end actually, and elasticsearch (our output) may fail to accept some push, and the whole pipeline gets stuck for 15 minutes until something somewhere times out and retries and everything deblocks? not sure how that would trigger a partition rebalance, because that timing from the first logs is still very suspicious, and two separate events matching up like that is very unlikely I think

I'll keep investigating, still very open to suggestions of things that could cause exactly 15 minute stalls / timeouts

[JoranDox]

current status: forked and changed this (with help from claude as I've never touched go before):

claude's words:

• Add the net import (needed for net.Dialer):

  import (
      "net"
      "net/http"
      ...

• Extend the http.Transport{} struct literal in esConfigFromParsed() — from:

  conf.clientOpts.Transport = &http.Transport{
      TLSClientConfig: tlsConf,
  }

to:

  conf.clientOpts.Transport = &http.Transport{
      TLSClientConfig:       tlsConf,
      DialContext:           (&net.Dialer{Timeout: 10 * time.Second}).DialContext,
      TLSHandshakeTimeout:   10 * time.Second,
      ResponseHeaderTimeout: 30 * time.Second,
      ExpectContinueTimeout: 1 * time.Second,
      IdleConnTimeout:       90 * time.Second,
  }

will report back if it works and if it doesn't I guess