Pending Events metric is always wrong

All the dependent JIRAs are resolved. Resolving this.

So this metric is implemented and the PR is out (see https://cdap.atlassian.net/browse/CDAP-2259#icft=CDAP-2259). Now I have one doubt: can this metric be queried as a time series? It seems that the metric system will only allow us to query the total value as an aggregate, but not over time. But that is what we would like to show in UI - how the queue size (pending events in the queue) is evolving. And if it keeps going up, that is an indication to scale out the flowlet to more instances.

@Alex Baranau can you advise how to do that best?

OK let's try to get this done using a metric in 3.0.

I have a suggestion on how to do it transactionally.

• For queue producer, along with enqueuing entries, increment a value per partition (basically groupId + instanceId) and store them in a separate row in the queue table (prefix with something unique to avoid conflict with queue entries).

  • This can records the number of entries enqueue for each partition

  • We can use readless increment to make the impact on performance minimal

  • Since writing to queue table is transactional, the metrics is always correct

• Similarly for queue consumer, each consumer performs a readless increment on number of dequeued entries, which get stored in the same row, but different column as the one used by the producer mentioned above.

  • The performance impact should be small, since it's just an extra write to HBase in the same batch (the queue consumer is already doing write to queue table to record the CONSUMED state).

  • Queue table is transactional, hence the metrics is always correct

• To query for the queue pending metrics, simply subtract the sum of all consumer metrics from the sum of all producer metrics

  • The sums can be computed by a single transactional get of the metric row in the queue table.

The change I mentioned above, however, is risky to be done in 3.0 timeframe (it involves modifying queue producer, consumer as well as queue eviction coprocessor). I think it make sense to use metrics system as an intermediate solution to keep pending metrics closer to what it is in 3.0 first.

Here is my suggested design:

• We add a new metric for each queue. Since each queue name already has the producing and consuming flowlet in its name, this will be a unique metric for each consumer group (flowlet) of a queue.

• The queue producer emits increments for this metric

• The queue consumer emits decrements for this metric

• When we clear the queues (reset queue state for a flow of for an app or namespace), we reset the value of the metric 0.

• Hence this metric always represents the number of unconsumed events per consumer, and the UI can query that.

We need to make sure that the increments and decrements are emitted exactly once for each event. That can be a little tricky, because metrics emission is not transactional.