[labrenbe]

Description

Add a local kafka-admin crate providing a Kafka client based on kafka-protocol, tokio and rustls that will be used by the Kafka agent.

Supported functions by the client:

• describe topics
• create topic
• create partitions (to scale up partitions)
• incremental alter (topic) configs
• delete topic

Definition of Done Checklist

• Not all of these items are applicable to all PRs, the author should update this template to only leave the boxes in that are relevant
• Please make sure all these things are done and tick the boxes

Author

• Changes are OpenShift compatible
• CRD changes approved
• CRD documentation for all fields, following the style guide.
• Helm chart can be installed and deployed operator works
• Integration tests passed (for non trivial changes)
• Changes need to be "offline" compatible
• Links to generated (nightly) docs added
• Release note snippet added

Reviewer

• Code contains useful comments
• Code contains useful logging statements
• (Integration-)Test cases added
• Documentation added or updated. Follows the style guide.
• Changelog updated
• Cargo.toml only contains references to git tags (not specific commits or branches)

Acceptance

• Feature Tracker has been updated
• Proper release label has been added
• Links to generated (nightly) docs added
• Release note snippet added
• Add type/deprecation label & add to the deprecation schedule
• Add type/experimental label & add to the experimental features tracker

[NickLarsenNZ]

Note to self, check if we enable the LogTracer (to get the logs). In 0.24, rustls will enable tracing so we won't need a log bridge for that.

[NickLarsenNZ]

nit: We should avoid positional (unnamed) interpolations.

Something like this reads nicer:

Suggested change

	if self.host.contains(':') {
	write!(f, "[{}]:{}", self.host, self.port)
	} else {
	write!(f, "{}:{}", self.host, self.port)
	}
	let Self { host, port } = self;
	if self.host.contains(':') {
	write!(f, "[{host}]:{port}")
	} else {
	write!(f, "{host}:{port}")
	}

[NickLarsenNZ]

To be even more sure, we could check by parsing the ipv6 address instead of looking for :.

[adwk67]

If tls::client_config is resolved once in KafkaAdminClient::connect, before the loop over the bootstrap servers, then a bad CA path fails there instead of surfacing as a Connect error for each server, and the PEM files are only read once per reconnect.

[adwk67]

Move the broker feature to dev-dependencies if/as it is only used for testing.

[adwk67]

check_version could be exposed as a pub(crate) method on Connection and called in KafkaAdminClient::send before Connection::send (dropping the check inside Connection::send), so that a failed version check does not throw away a healthy connection. e.g.

// client.rs
async fn send<R: Request>(&mut self, request: R, version: i16) -> Result<R::Response> {
    // return both here so that last_used can be used when putting the connection back into the pool
    let (mut connection, last_used) = match self.connection.take() {
        Some((connection, last_used)) if last_used.elapsed() < CONNECTION_MAX_IDLE => {
            (connection, last_used)
        }
        _ => (self.connect().await?, Instant::now()),
    };

    // The check doesn't do any I/O, so the connection stays usable and can go back in the pool.
    if let Err(source) = connection.check_version(R::KEY, version) {
        self.connection = Some((connection, last_used));
        return Err(Error::Send { source });
    }

    let response = connection.send(request, version).await.context(SendSnafu)?;
    self.connection = Some((connection, Instant::now()));
    Ok(response)
}