add Core consumer impl; capable of processing messages from a consumer client.

[gauravAshok]

• netty executor group.
• Is consumer tied to a context? Is consumer a "task" to be run on a event-loop.
• Consumer initialization:
  • blocking init() / connect()
  • startOn(EventLoop)

---

Consumer task? (ctx(event_loop))

• MessageSrc::replenish : Launch a task to fetch new messages from outside world & then enqueue a task on ctx to replenish this messageSrc.
  • async MessageSrc::nextMessages(out array) : async method to fetch messages from the messageSrc. It is a wrapper to hide a particular consumer impl and grouping semantics. It allows us to reorder (if required) messages received from the consumer instance.
• process LOOP : loop_begin:
  • async nextMessages() : fetch messages from the messageSrc.
  • then, check for group state
  • and, split(messages) into (failed, ok) messages
  • and, enqueue concurrency_control.deliver(ok_message)
  • and, enqueue produce_quota_q.acquire(size(failed_msgs)), // this step is simplified, but each message can go to a separate q, hence the quota object may be multiple
  • then enqueue loop_begin.
• ConcurrencyControl (a Thread safe DS)
  • deliver(msgs)
    • requests_to_make = min(free_slots, msgs.size)
    • add_backlog(rest of the messages)
    • free_slots -= requests_to_make
    • msgs.slice(reqeusts_to_make).forEach(msg -> enqueue_delivery(msg)), then
      • free_slots += 1
      • and if requests_in_backlog, then add a task to initiate delivery for them onto the event_loop, if the task not already started.
      • and, enqueue post_process(msg, response) on ctx
• Post_process(msg, response): This is running back on ctx / event-loop.
  • If it failed, then enqueue error_throttler.acquire(produce_to_failure_q)
    • then, q.produce()
    • then, GS state upte.
    • msg.commit()
  • else, advance GS state and msg.commit()

todo: explore if post_process needs to be batched!, everything before the post_process is batched upto some extent. post_process is the only step that is per msg. enqueue from outside onto the ctx, is also a candidate for contention. Should the event-loop be responsible for http/remote requests? probably not, as one is highly IO bound and other is compute bound. IO threads need to be responsive.

[aayustark007-fk]

How do we handle Consumer failure scenarios and prevent duplicate delivery during recovery? Eg: Consumer task fails post ConcurrencyControl::deliver for some of the messages. So, the offset is not committed to the messaging stack. Now, when the consumer comes back up, or the workload is assigned to another consumer. It will try to deliver those messages again. Are we fine with that?

[gauravAshok]

No progress on it yet. expected to pick this up on 28th oct.

[anuj-flipkart]

Pending:

1. Stopping of consumer.
2. Async HTTP Client implementation
3. Testing

[gaurav-ashok]

Testing in progress. E2E consumption is happening.

2024-11-20 04:43:58 INFO  ProcessingLoop:108 - Delivered message. status: 200
2024-11-20 04:43:58 INFO  ProcessingLoop:108 - Delivered message. status: 200
2024-11-20 04:43:58 INFO  ProcessingLoop:108 - Delivered message. status: 200
2024-11-20 04:43:58 ERROR ProcessingLoop:75 - unexpected error in fetching messages from msgSelector
java.lang.IllegalStateException: Only one request is allowed at a time
    at com.flipkart.varadhi.consumer.MessageSrcSelector.nextMessages(MessageSrcSelector.java:48) ~[classes/:?]
    at com.flipkart.varadhi.consumer.processing.ProcessingLoop.run(ProcessingLoop.java:72) ~[classes/:?]
    at com.flipkart.varadhi.consumer.concurrent.EventExecutor.runSafely(EventExecutor.java:81) ~[classes/:?]
    at com.flipkart.varadhi.consumer.concurrent.EventExecutor.run(EventExecutor.java:68) ~[classes/:?]
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.111.Final.jar:4.1.111.Final]
    at java.base/java.lang.Thread.run(Thread.java:1583) [?:?]

But looks like there some invariants being violated. Need to debug them.