Concurrent access issues with plugins

Using go routine to frequently call this method causes the service to crash.

package main
import (
"context"

"github.com/benthosdev/benthos/v4/public/service"

// Import only required Benthos components, switch with `components/all` for
// all standard components.
_ "github.com/benthosdev/benthos/v4/public/components/io"
_ "github.com/benthosdev/benthos/v4/public/components/pure"

)

func main() {
panicOnErr := func(err error) {
if err != nil {
panic(err)
}
}

builder := service.NewStreamBuilder()

// Set the full Benthos configuration of the stream.
err := builder.SetYAML(`

input:
generate:
count: 1
interval: 1ms
mapping: 'root = "hello world"'

processors:
- mapping: 'root = content().uppercase()'

output:
stdout: {}

logger:
level: none
`)
panicOnErr(err)

// Build a stream with our configured components.
stream, err := builder.Build()
panicOnErr(err)

// And run it, blocking until it gracefully terminates once the generate
// input has generated a message and it has flushed through the stream.
err = stream.Run(context.Background())
panicOnErr(err)

}

How to solve this problem?

Start with the provided Go example and inspect the service.NewStreamBuilder and stream.Run entry points, then reproduce the crash under concurrent access. Identify the relevant plugin or service lifecycle behavior and establish a regression test; done means repeated concurrent calls no longer crash the service.