Implement Event-Driven Cache Invalidation for Producer State Synchronization

[devanshu-garg-IN]

Description

Currently, the producer service maintains an internal cache where topics are lazily fetched and stored with an expiry time of 3600 seconds from the last access. However, there is an issue with state progression across all pods, leading to outdated cache entries.

To address this, we will implement an event-driven mechanism where a controller propagates entity updates (create/update/delete) to all services that maintain this cache. This will ensure that state changes are accurately reflected across all pods in real time.

Acceptance Criteria

1. Event Handling in All Services

• Implement an EventHandler in the REST layer.

• The event structure will be:

  {
    "resource_type": "Topic/Subscription/...",
    "resource_name": "",
    "type": "created/updated/deleted",
    "body": {}
  }

• The body will be null in case of deletion.

2. Event Registration & Delivery

• All services should register themselves in the event-bus so that the controller can identify recipients.
• The controller should push events to all registered services when a state change occurs.

3. Zookeeper Integration for Event Propagation

• Any entity create/update/delete in Zookeeper should trigger a corresponding MarkerEntity in Zookeeper via a transactional operation.
• The controller listens to this path and propagates the event to all pods along with the resource payload.

4. Efficient Event Propagation

• Use Java 21 Virtual Threads to spawn a lightweight thread per pod that continuously pushes events in a loop.

5. Monitoring & Metrics

• Emit metrics to track event propagation lag per pod to monitor system health and detect bottlenecks.