Discard duplicate `BrokerMessage` instances (adding idempotent behaviour)

[garciparedes]

Issue by garciparedes
Monday Dec 27, 2021 at 11:41 GMT
Originally opened as Clariteia/minos_microservice_networks#472

---

The proposed solution is highly inspired by: https://chrisrichardson.net/post/microservices/patterns/2020/10/16/idempotent-consumer.html

Here is a class diagram containing a proposal to include the IdempotentBrokerSubscriber following a Chain of Reponsability design pattern:

classDiagram
    

    class BrokerSubscriber {
        <<abstract>>
        _topics: set[str]
        set[str] topics()
        receive() BrokerMessage
        _receive()* BrokerMessage
    }


    BrokerSubscriber <|-- KafkaBrokerSubscriber
    
    class KafkaBrokerSubscriber {
        broker_host: str
        broker_port: int
        _from_config(config: MinosConfig) KafkaBrokerSubscriber
        _setup()
        _destroy()
        _create_topics()
        _delete_topics()
        _receive(message: BrokerMessage)
        client: AIOKafkaConsumer
        admin_client: KafkaAdminClient
    }


    BrokerSubscriber <|-- QueuedBrokerSubscriber
    
    class QueuedBrokerSubscriber {
        _run_task: Optional[asyncio.Task]
        _setup()
        _destroy()
        _start_task()
        _run()
        _stop_task()
        _receive(message: BrokerMessage)
    }
    
    QueuedBrokerSubscriber "1" o-- "1" BrokerSubscriber : impl
    QueuedBrokerSubscriber "1" o-- "1" BrokerSubscriberQueue : queue

    class BrokerSubscriberQueue {
        <<abstract>>
        _topics: set[str]
        set[str] topics()
    }

    BrokerSubscriber <|-- IdempotentBrokerSubscriber

    class IdempotentBrokerSubscriber {
        _setup()
        _destroy()
        _receive(message: BrokerMessage)
    }
    
    IdempotentBrokerSubscriber "1" o-- "1" BrokerSubscriber : impl
    IdempotentBrokerSubscriber "1" o-- "1" BrokerSubscriberDuplicateDetector : duplicate_detector

    class BrokerSubscriberDuplicateDetector {
        <<abstract>>
        is_duplicate(topic: str, uuid: UUID)* bool
    }

Loading