Support precreated Kafka topics on locked-down brokers

[elindsey]

Some customers would like to run materialized in a configuration where topics are precreated in Kafka, the broker disallows materialized from creating new topics, and materialize's sinks are created with reuse_topic.

This is currently unsupported.
Our logic today is:

1. Attempt to create topic
2. If we receive a TopicAlreadyExists error, consider creation successful
3. Do a metadata query to confirm that the created topic matches what we expect

In a scenario where the Kafka broker has disallowed topic creation, step 1 will result in a denied error - even if the topic exists.
Example:

ERROR: error registering kafka topic for sink: error creating topic foo-sink for sink: TopicAuthorizationFailed (Broker: Topic authorization failed)

Logic will need to be adjusted to something that both handles the topic creation race condition and supports locked-down brokers, such as:

1. Metadata query to check for topic existence
2. If doesn't exist, attempt to create topic
3. If we receive a TopicAlreadyExists error, consider creation successful
4. Do a metadata query to confirm that the created topic matches what we expect

It will be important to test this in a plausible user configuration to make sure that whatever we put in as that initial metadata query isn't also going to be ACL denied.

[elindsey]

Note that there are two places in code where creation happens: register_kafka_topic in the sink_connector and create_topic in the kafka-util admin.rs.

As part of this task, we should refactor it a bit so this logic can live in one place. One option might be having the sink_connector use our kafka-utils.

[quodlibetor]

we should probably have the same semantics for that as we end up implementing the sqs part of s3 sources with sqs notifications.

That is: similar syntax for using existing vs creating new, similar defaults, similar error and cleanup behavior.

[ruf-io]

Heroku Kafka also works this way.

So, even when topics are pre-created by the user, attempting to create SINKs to Heroku Kafka fails at step (1) with the error:

ERROR:  error registering kafka topic for sink: Error creating topic highvalusers for sink: Admin operation error: TopicAuthorizationFailed (Broker: Topic authorization failed): Admin operation error: TopicAuthorizationFailed (Broker: Topic authorization failed)

Not too sure about the popularity of Kafka on Heroku, but just flagging it here.

[elindsey]

Note - it’s possible that the metadata check in precreated topics should differ from the metadata query when mz creates them (for example, does the sink actually have a dependency on knowing the partition count for data topics or can it just use whatever topic is present however it’s configured).

[sploiselle]

Retiring old issue