Recovery from "all broker connections are down"

Hi There,

During some performance testing done with a ReplicaSet for applications hosted in Azure Kubernetes Service, one of the pods encountered this issue:


`{ message: [LibrdKafkaError: Local: All broker connections are down] { origin: 'local', message: 'all broker connections are down', code: -1, errno: -1 }, name: '#producer-4', fac: 'BINDING', timestamp: 1739269165001 }`

The producer is instantiated through a Singleton, below is the code I have changed/redacted due to sensitivity issues. It will use SASL_SSL for authentication when running on the cluster and if not Auth config is passed in it will not use any Auth (used for local testing where broker is hosted in docker container)

```
import { KafkaJS } from '@confluentinc/kafka-javascript';
import { IKafkaAuthConfig } from './IKafkaAuthConfig';
import axios from 'axios';
import KafkaLogger from '../utils/KafkaLogger';

const logger: KafkaLogger = KafkaLogger.getInstance();

/**
* Singleton class for Kafka Producer
*/
class KafkaProducer {
private static instance: KafkaProducer;

private kafkaBootstrapURL: string
private kafkaAuthConfig: IKafkaAuthConfig | undefined;
private kafkaLogger: KafkaJS.Logger;

private constructor(kafkaBootstrapURL: string, kafkaAuthConfig?: IKafkaAuthConfig) {
this.kafkaBootstrapURL = kafkaBootstrapURL
this.kafkaAuthConfig = kafkaAuthConfig
this.kafkaLogger = logger
}

/**
* Get instance of KafkaProducer if not already instantiated
*
* @param kafkaBootstrapURL - Required URL for both Cloud and Local
* @param kafkaConfig - Required for AKS integration for SASL authentication
* @returns KafkaJS.Producer
*/
public static async getInstance(
kafkaBootstrapURL: string,
kafkaAuthConfig?: IKafkaAuthConfig
): Promise {
if (!KafkaProducer.instance) {
KafkaProducer.instance = new KafkaProducer(kafkaBootstrapURL, kafkaAuthConfig)
}
return KafkaProducer.instance;
}

/**
* Returns a producer for confluence Kakfa using the parameters in the instance
* If kafkaAuthConfig is not defined, then it is assumed that this is connecting to a local
* instance (e.g. docker) where SASL is not required.
*
* For integration between AKS deployed applications and Confluent Cloud, the kafkaAuthConfig contains
* the OAuth parameters for use with the application's service principal
* Connect is called by the application that gets the instance of KafkaProducer
* @returns KafkaJS.Producer
*/
public async getProducer(): Promise {
try {
let producer: KafkaJS.Producer;
if (!this.kafkaAuthConfig) {
const kafka = new KafkaJS.Kafka({
kafkaJS: {
brokers: [this.kafkaBootstrapURL],
clientId: `app-name`,
logger: this.kafkaLogger
}
})
producer = kafka.producer();
} else {
const kafka = new KafkaJS.Kafka({
kafkaJS: {
brokers: [this.kafkaBootstrapURL],
clientId: ``,
ssl: true,
sasl: {
mechanism: 'oauthbearer',
oauthBearerProvider: this.tokenRefresh.bind(this)
},
logger: this.kafkaLogger
}
})
producer = kafka.producer();
}
return producer;
} catch (error: any) {
console.error(error);
throw new Error(error);
}
}

/**
* Callback function for oauthBearerProvider to manage OAuth and token refresh.
*
* It retrieves a token from the OAuth endpoint using the Confluent Cloud scope, extracts
* out the expiry time for the Kafka package to handle the refresh time (80% of expiry) and
* sets the other parameters such as the logicalCluster and identityPoolId
*
* @returns { value: token, lifetime: exp_ms, principal, extensions };
*/
private async tokenRefresh(): Promise {
if (!this.kafkaAuthConfig) {
throw new Error('No Kafka configuration Provided for Token Refresh')
}
try {
const response = await axios.post(this.kafkaAuthConfig.kafkaOAuthTokenEndpointURL, new URLSearchParams({
grant_type: 'client_credentials',
client_id: this.kafkaAuthConfig.azureClientSPN,
client_secret: this.kafkaAuthConfig.azureClientSecret,
scope: this.kafkaAuthConfig.kafkaOAuthTokenScope
}), {
headers: {
'Content-Type': 'application/x-www-form-urlencoded'
}
});

// Extract the token and expiration time from the response
const token = response.data.access_token;
const exp_seconds = Math.floor(Date.now() / 1000) + response.data.expires_in;
const exp_ms = exp_seconds * 1000;

const principal = 'admin';
const extensions = {
logicalCluster: this.kafkaAuthConfig.kafkaLogicalCluster,
identityPoolId: this.kafkaAuthConfig.kafkaIdentityPool
};

return { value: token, lifetime: exp_ms, principal, extensions };
} catch (error: any) {
console.error(`Failed to retrieve OAuth token: ${error.message}`);
throw new Error(error.message);
}
}
}

export default KafkaProducer;
```

The logged exception was not caught by the catch blocks and the producer never recovered. How is this situation meant to be handled based upon the features provided in the library?

Start with the shown KafkaProducer class, especially getProducer and the producer connect/send lifecycle around the reported LibrdKafkaError. Trace how connection failures are surfaced outside the shown try/catch blocks, then verify the library's documented recovery behavior after broker connections return. Done means the handling is documented or reproduced with a focused test, including the SASL_SSL and local configurations described.

Written by the indexing model from the issue text.