Initial Kafka support

[hartungstenio]

How to use:

class ProtobufSchemaMessageTranslator(KafkaMessageTranslator):
    def __init__(self, message_type):
        self.deserializer = ProtobufDeserializer(message_type)

    def translate(self, message):
        translated = super().translate(message)

        ctx = SerializationContext(translated["metadata"]["topic"], "value")
        translated["content"] = self.deserializer(translated["content"], ctx)

        return translated

routes = (
    KafkaRoute(
        "my_topic",
        provider_options,
        handler=Handler(),
        message_translator=ProtobufSchemaMessageTranslator(my_schema),
    ),
)

manager = LoaferManager(routes)
manager.run()

[hartungstenio]

O Provider só aceita uma configuração, e uso ela para produtores e consumidores.
A maioria dos parâmetros é compartilhada entre eles. Essa parada do inspect é para eu conseguir separar as poucas diferenças

[hartungstenio]

Estou fazendo o commit manualmente quando consigo puxar mensagens, então não quero o auto commit.

[hartungstenio]

O Kafka (e coisas parecidas) não possuem o conceito de confirmar uma mensagem. Eu consigo confirmar até onde eu já li.

Então a ideia aqui é que sempre que o handler ou o error_handler retornarem False, eu republico a mensagem novamente no Kafka.

[hartungstenio]

Metadado de quantas tentativas já foram.

[hartungstenio]

Não gostei muito dessa lógica, mas ainda não pensei em algo melhor.

A ideia aqui é: Como sempre republico a mensagem, quero aqui falar em qual tópico vou republicar. Então isso é uma função que recebe o tópico original, e a quantidade de tentativas.

Assim consigo criar uma "DLQ" no Kafka. Algo como:

lambda topic, tries: topic if tries < 5 else f"dead__{topic}"