...

The challenge with supporting this additional feature is that the "min.insync.replicas" configuration may be set at a broker or topic level. We can get the configured

We can use the same technique `LogManager` uses to determine the "min.insync.replicas" for a given topic by:

(1) Check Get broker/cluster-level configuration (from `AdminClient.describeConfigs`)

(2) Get topic-level configuration in Zookeeper(from `KafkaZkClient.getLogConfigs` and passing in default "min.insync.replicas" as the value retrieved from (1))

We must pass the default value as a parameter in (2) as `getLogConfigs` will return a default value if the specified configuration is not specifically set in Zookeeper (Kafka default for "min.insync.replicas" is 1). If the user has configured "min.insync.replicas" on the broker/cluster-level (and not topic-level), then `getLogConfigs` will return the default value so we must make sure it has the value used by the broker/cluster.(2) Use AdminClient to get broker/cluster-level configuration

This means we must add an additional flag "--bootstrap-server" to use AdminClient to describe broker configurations when we cannot find the configuration override get the broker/cluster "min.insync.replicas" configuration to use as the default when it is not found in Zookeeper.

Compatibility, Deprecation, and Migration Plan

...