confluentinc / confluentinc/confluent-kafka-javascript

Bug: Connecting multiple consumers / producer with "sasl: oauthbearer" in Promise.all fails with "Error Code -172: Local: Erroneous state"

Open
#254 5 comments 2 reactions 0 assignees View on GitHub
bug
Dominant language
TypeScript
Stars
304
Forks
45
Avg merge
11h 47m
Merged PRs (30d)
5

Description

**Environment Information**
- OS [e.g. Mac, Arch, Windows 10]: Linux on arm64
- Node Version [e.g. 8.2.1]: 20
- NPM Version [e.g. 5.4.2]: `pnpm@9.15.1`
- C++ Toolchain [e.g. Visual Studio, llvm, g++]: n/a
- confluent-kafka-javascript version [e.g. 2.3.3]: 1.2.0

**Steps to Reproduce**
1. Create multiple instances of a `KafkaJS.Producer` or `KafkaJS.Consumer`, with the following configuration in the `KafkaJS.Kafka` client:
```typescript
{
kafkaJS: {
brokers: ["my_aws_brokers:9098"], // config value
clientId: "singleton-client",
logger: new KafkaCustomLogger(),
logLevel: logLevel["WARN"], // config value
retry: {
maxRetryTime: 30_000, // config value
retries: 500, // config value
},
ssl: true,
sasl: {
mechanism: "oauthbearer" as const,
oauthBearerProvider: async () => {
const authTokenResponse = await generateAuthToken(); // from aws-msk-iam-sasl-signer-js
return {
value: authTokenResponse.token,
lifetime: authTokenResponse.expiryTime,
principal: "",
};
}
}
}
```
- Note: The same thing happens with multiple instances of a `KafkaJS.Kafka` client, or when there is a single client.
2. Call `await consumer.connect()` or `await producer.connect()` for each of these instances in a `Promise.all`
3. The following errors appear:
- `KafkaJSConnectionError: broker transport failure`, associated type is `ERR__TRANSPORT`, error code `-195`
- `Local: Erroneous state`, error code `-172`
- `[thrd:sasl_ssl://{my_aws_broker}]: sasl_ssl://b-3.{my_aws_broker}:9098/bootstrap: SASL authentication error: SaslAuthenticateRequest failed: Local: Broker handle destroyed (after 0ms in state DOWN)`

**confluent-kafka-javascript Configuration Settings**
Shared above

**Additional context**

I believe this is related to the `oauthbearer` mechanism specifically, as I am able to successfully connect using an unauthenticated connection in a `Promise.all`. However, our production clusters utilize `IAM` auth, so this wouldn't work for us there.

Note that I am able to successfully connect using `oauthbearer` if I connect in a for loop, like:

```typescript
for (const kafkaClient of kafkaClients) {
await kafkaClient.consumer(consumerOptions).connect();
await kafkaClient.producer(producerOptions).connect();
}
```

We will likely move forward using this pattern, but it's an unexpected result that a `Promise.all` would cause this issue.

For additional context, we are using `Promise.all` as we are initializing a set of different Nest applications simultaneously. The consumer and producer connections occur in the initialization of each of these independent applications.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.