confluentinc / confluentinc/confluent-kafka-javascript
Bug: Connecting multiple consumers / producer with "sasl: oauthbearer" in Promise.all fails with "Error Code -172: Local: Erroneous state"
- 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
Assessment
This issue has not been assessed yet.