Calculate offset lag for test expectation

A test checks the offset lag of consumers (last committed offset -
consumer position). The test needs to have the number of received
messages to calculate the expected offset lag, it cannot be a fixed
value like previously.

Author: acogoluegnes

deps/rabbitmq_prometheus/test/rabbit_prometheus_http_SUITE.erl

@@ -805,11 +805,13 @@ stream_pub_sub_metrics(Config) ->
     ct:pal("Initial metrics: ~p", [Metrics]),
 
     Stream1 = atom_to_list(?FUNCTION_NAME) ++ "1",
-    MsgPerBatch1 = 2,
-    {ok, S1, C1} = publish_via_stream_protocol(list_to_binary(Stream1), MsgPerBatch1, Config),
+    MsgPerBatch1 = 2, %% we'll publish 2 batches
+    {ok, S1, C1, MsgCount1} = publish_via_stream_protocol(list_to_binary(Stream1),
+                                                          MsgPerBatch1, Config),
     Stream2 = atom_to_list(?FUNCTION_NAME) ++ "2",
-    MsgPerBatch2 = 3,
-    {ok, S2, C2} = publish_via_stream_protocol(list_to_binary(Stream2), MsgPerBatch2, Config),
+    MsgPerBatch2 = 3, %% we'll publish 2 batches
+    {ok, S2, C2, MsgCount2} = publish_via_stream_protocol(list_to_binary(Stream2),
+                                                          MsgPerBatch2, Config),
 
     %% aggregated metrics
 
@@ -825,12 +827,15 @@ stream_pub_sub_metrics(Config) ->
 
     %% per-object metrics
      {_, Body2} = http_get_with_pal(Config, "/metrics/detailed?family=stream_consumer_metrics",
-                                   [], 200),
+                                    [], 200),
     ParsedBody2 = parse_response(Body2),
     #{rabbitmq_detailed_stream_consumer_max_offset_lag := MaxOffsetLag} = ParsedBody2,
 
-    ?assertEqual([{#{vhost => "/", queue => Stream1}, [2]},
-                  {#{vhost => "/", queue => Stream2}, [3]}],
+    %% we published 2 batches and received a first chunk
+    ExpectedLag1 = MsgPerBatch1 * 2 - MsgCount1,
+    ExpectedLag2 = MsgPerBatch2 * 2 - MsgCount2,
+    ?assertEqual([{#{vhost => "/", queue => Stream1}, [ExpectedLag1]},
+                  {#{vhost => "/", queue => Stream2}, [ExpectedLag2]}],
                  lists:sort(maps:to_list(MaxOffsetLag))),
     dispose_stream_connection(S1, C1, list_to_binary(Stream1)),
     dispose_stream_connection(S2, C2, list_to_binary(Stream2)),
@@ -926,8 +931,13 @@ publish_via_stream_protocol(Stream, MsgPerBatch, Config) ->
     {ok, C6} = stream_test_utils:subscribe(S, C5, Stream, SubscriptionId, _InitialCredit = 0),
     ok = stream_test_utils:credit(S, SubscriptionId, 1),
     %% delivery of first batch of messages
-    {{deliver, SubscriptionId, _Bin1}, C7} = stream_test_utils:receive_stream_commands(S, C6),
-    {ok, S, C7}.
+    {{deliver, SubscriptionId, Bin1}, C7} = stream_test_utils:receive_stream_commands(S, C6),
+    <<5:4/unsigned,
+      0:4/unsigned,
+      _ChType:8/unsigned,
+      NumEntries:16/unsigned,
+      _Rest/binary>> = Bin1,
+    {ok, S, C7, NumEntries}.
 
 dispose_stream_connection(Sock, C0, Stream) ->
     {ok, C1} = stream_test_utils:delete_stream(Sock, C0, Stream),