feat(buffer_worker): refactor buffer/resource workers to always use queue and use offload mode

[thalesmg]

https://emqx.atlassian.net/browse/EMQX-8623

Currently, we face several issues trying to keep resource metrics
reasonable. For example, when a resource is re-created and has its
metrics reset, but then its durable queue resumes its previous work
and leads to strange (often negative) metrics.

Instead using counters that are shared by more than one worker to
manage gauges, we introduce an ETS table whose key is not only scoped
by the Resource ID as before, but also by the worker ID. This way,
when a worker starts/terminates, they should set their own gauges to
their values (often 0 or replayq:count when resuming off a queue).
With this scoping and initialization procedure, we'll hopefully avoid
hitting those strange metrics scenarios and have better control over
the gauges.

This makes the buffer/resource workers always use replayq for
queuing, along with collecting multiple requests in a single call.
This is done to avoid long message queues for the buffer workers and
rely on replayq's capabilities of offloading to disk and detecting
overflow.

Also, this deprecates the enable_batch and enable_queue resource
creation options, as: i) queuing is now always enables; ii) batch_size > 1
<=> batch_enabled. The corresponding metric
dropped.queue_not_enabled is dropped, along with batching. The
batching is too ephemeral, especially considering a default batch time
of 20 ms, and is not shown in the dashboard, so it was removed.

Also, fixes a bug related to message loss in kafka producer when connection is down. Currently, the kafka producer bridge will test the connection to kafka itself to say if the resource is connected. However, if kafka or its connection is down, this’ll make messages to be lost, as there are no buffer workers for this bridge, and, being “down”, the resource won’t call the wolff producers, leading to both message loss and wrong metrics (as there won’t be failed counters bumped).

[thalesmg]

Based/depends on #9619

[thalesmg]

Matched moved here because otherwise retries would bump the matched metric as if they were new requests.

[thalesmg]

Changed to pop because, previously, the head of the queue would be duplicated N times.

[thalesmg]

Fixme: the telemetry handler ID is changed everytime the resource is created, leading to duplicate metrics.

[lafirest]

Maybe the block/2 also needs to add this line, emqx_resource_metrics:matched_inc(Id, length(Query)), ?

[thalesmg]

Makes sense. But searching the code base, I could not find any usage of block/2 🤔

I think I'll just remove it. 😺

[lafirest]

There may be a bug if data is in the replayQ, its HasBeenSent will always be false even though it may have been sent N times

[thalesmg]

Before this refactoring, the only requests that did bump the retried.* counters were the async requests, and they are the only ones that do get inserted into the inflight table and marked as sent.

I'm not entirely sure this was the intended behavior indeed. But I think that this behavior is kept because, after this refactoring, the path that an async request does is:

1. The request gets enqueued into replayq.
2. It's then popped (possible in a batch).
3. It's appended to the inflight table (if there's room).
4. In the first ("outer") handle_query_result, since it'll return {async_return, ok}, it'll be ack'ed in replayq, removing it from there.
5. Eventually, when {batch_,}reply_after_query is called, then it'll bump either retried.* counter.

For sync requests, you are right: they won't bump the retried.* counters. Do you think that's the original intention? We would need to track what has been retried somehow.

We can't change the replayq items without re-appending them to the queue, and that would change the order of the requests, unfortunately. One way could be to keep a table of hashes of requests and then check if those have been sent or not. 🤔

I'm assuming that we want to keep the order in replayq to prevent things like re-ordering messages from the same client.

What do you think?

[lafirest]

Personally, I think we can just ignore this here, then create a track ticket to the product owner, and left the discussion to them

[thalesmg]

Sounds good to me: we can revisit this point in a future PR, if required. 😺