feature: [VRD-679] New client function get_kafka_topics()

[ewagner-verta]

Impact and Context

• Adds a new function to verta.client.Client that will allow users to fetch a list of available Kafka topics.
• Updates KafkaSettings class to make the cluster_config_id argument optional.
• Refactors verta._endpoint.Endpoint() to fetch the Kafka config ID when it is not included.
• Adds property/unit tests to verify behavior with and without cluster_config_id.
• Adds property/unit tests for new client function.
• Includes some minor black formatting changes to bring changed files up to standard.

Risks and Area of Effect

Testing

• Unit test
• test_endpoint.py

• test_client_get_kafka_topics.py

• Deployed to dev env
• Other (explain)
  Validation script for running against a real env with a Kafka config:

from verta import Client
from verta.endpoint import KafkaSettings
from verta.environment import Python
from verta.registry import verify_io, VertaModelBase


class Model(VertaModelBase):
    def __init__(self, artifacts=None):
        pass

    @verify_io
    def predict(self, input):
        return input


def main():
    client = Client()

    model_ver = client.get_or_create_registered_model(
        f"Kafka Model Testing"
    ).create_standard_model(
        Model,
        code_dependencies=[],
        environment=Python([]),
    )

    kafka_settings = KafkaSettings(
        input_topic="inputTopic",
        output_topic="outputTopic",
        error_topic="errorTopic",
    )
    assert kafka_settings._cluster_config_id is None
    print(f"\nKafka Settings Provided\n----------------------\n{kafka_settings}\n")

    client.create_endpoint("kafka-model-testing", kafka_settings=kafka_settings)
    endpoint = client.get_endpoint("kafka-model-testing")
    assert endpoint.kafka_settings.cluster_config_id is not None

    print(f"\nFinal Kafka Settings\n--------------------\n{endpoint.kafka_settings}\n")

if __name__ == "__main__":
    main()

Output:

got VERTA_HOST from environment
got VERTA_EMAIL from environment
got VERTA_DEV_KEY from environment
connection successfully established
created new RegisteredModel: Kafka Model Testing in workspace: ewagner
created new ModelVersion: ModelVersion 421771680719193545579
uploading model to Registry
upload complete
uploading model_api.json to Registry
upload complete
uploading custom_modules to Registry
upload complete

Kafka Settings Provided
----------------------
KafkaSettings('inputTopic', 'outputTopic', 'errorTopic', None)

Final Kafka Settings
--------------------
KafkaSettings('inputTopic', 'outputTopic', 'errorTopic', 'd5a8b0b8-c4ab-4d75-97e7-ecddc494402f')

Note the cluster_config_id (4th arg) has been fetched.

Reverting

• Contains Migration - Do Not Revert

[ewagner-verta]

!rebuild

[liuverta]

There are past comments that weren't addressed. Are they fine?

• feature: [VRD-679] New client function get_kafka_topics() #3727 (comment)
• feature: [VRD-679] New client function get_kafka_topics() #3727 (comment)

[ewagner-verta]

@liuverta Did you happen to see my prior responses in each of the two threads you referenced?

In the first case, using assume to ensure uniqueness in a list as suggested is failing for me, so I'm asking if you can make a specific code suggestion there or can we leave it working as is?

In the second case, sending a None value to the back-end would result in a 500 error that I think is ok to pass along as is. Again, if you have a specific suggestion or vision for that process, can you share?

[liuverta]

Thanks for tackling the tests 😵‍💫 Some feedback on the error messages, then I think we'll be good

[ewagner-verta]

@liuverta - No problem. Thanks for the review. I accepted the changes to the error messages and I just pushed a commit to update the unit tests accordingly. Cheers