anchor.diff:  ?DEFINE_EXTRACT(backoff, queue_backoff);
   ?DEFINE_EXTRACT(backfill_remaining, queue_backfillremaining);
   ?DEFINE_EXTRACT(itemondisk, queue_itemondisk);
  +?DEFINE_EXTRACT(total_backlog_size, total_backlog_size);
   ?DEFINE_EXTRACT(count, count);
   extract_agg_stat(_K, _V, Acc) -> Acc.
   -undef(DEFINE_EXTRACT).
  
positive.code:                       ?DEFINE_SUBTRACTOR(queue_drain),
                        ?DEFINE_SUBTRACTOR(queue_backoff),
                        ?DEFINE_SUBTRACTOR(queue_backfillremaining),
                        ?DEFINE_SUBTRACTOR(queue_itemondisk)}.
  -undef(DEFINE_SUBTRACTOR).
  
  -endif. % NEED_TAP_STREAM_STATS_CODE
  
negative.code:                            queue_drain = 0,
                             queue_backoff = 0,
                             queue_backfillremaining = 0,
                             queue_itemondisk = 0,
                             total_backlog_size = 0}).
  
  -define(TAP_STAT_GAUGES,
          ep_tap_rebalance_count, ep_tap_rebalance_qlen, ep_tap_rebalance_queue_backfillremaining, ep_tap_rebalance_queue_itemondisk, ep_tap_rebalance_total_backlog_size,
          ep_tap_replica_count, ep_tap_replica_qlen, ep_tap_replica_queue_backfillremaining, ep_tap_replica_queue_itemondisk, ep_tap_replica_total_backlog_size,
          ep_tap_user_count, ep_tap_user_qlen, ep_tap_user_queue_backfillremaining, ep_tap_user_queue_itemondisk, ep_tap_user_total_backlog_size,
          ep_tap_total_count, ep_tap_total_qlen, ep_tap_total_queue_backfillremaining, ep_tap_total_queue_itemondisk, ep_tap_total_total_backlog_size).
  -define(TAP_STAT_COUNTERS,
          ep_tap_rebalance_queue_fill, ep_tap_rebalance_queue_drain, ep_tap_rebalance_queue_backoff,
          ep_tap_replica_queue_fill, ep_tap_replica_queue_drain, ep_tap_replica_queue_backoff,
  