Fix conflicts

References #7743

Author: acogoluegnes

deps/rabbit/src/rabbit_core_ff.erl

@@ -111,7 +111,14 @@
       depends_on    => [stream_queue]
      }}).
 
-<<<<<<< HEAD
+-rabbit_feature_flag(
+   {stream_sac_coordinator_unblock_group,
+    #{desc          => "Bug fix to unblock a group of consumers in a super stream partition",
+      doc_url       => "https://github.com/rabbitmq/rabbitmq-server/issues/7743",
+      stability     => stable,
+      depends_on    => [stream_single_active_consumer]
+     }}).
+
 %% -------------------------------------------------------------------
 %% Direct exchange routing v2.
 %% -------------------------------------------------------------------
@@ -231,13 +238,3 @@ delete_table(FeatureName, Tab) ->
             %% adheres to the callback interface
             ok
     end.
-=======
-
--rabbit_feature_flag(
-   {stream_sac_coordinator_unblock_group,
-    #{desc          => "Bug fix to unblock a group of consumers in a super stream partition",
-      doc_url       => "https://github.com/rabbitmq/rabbitmq-server/issues/7743",
-      stability     => stable,
-      depends_on    => [stream_single_active_consumer]
-     }}).
->>>>>>> 221f10d2d9 (Unblock group of consumers on super stream partition)

deps/rabbit/src/rabbit_stream_sac_coordinator.erl

@@ -274,11 +274,7 @@ apply(#command_activate_consumer{vhost = VirtualHost,
     {G, Eff} =
         case lookup_group(VirtualHost, Stream, ConsumerName, StreamGroups0) of
             undefined ->
-<<<<<<< HEAD
-                rabbit_log:warning("trying to activate consumer in group ~p, but "
-=======
                 rabbit_log:warning("Trying to activate consumer in group ~tp, but "
->>>>>>> 221f10d2d9 (Unblock group of consumers on super stream partition)
                                    "the group does not longer exist",
                                    [{VirtualHost, Stream, ConsumerName}]),
                 {undefined, []};
@@ -828,10 +824,8 @@ send_message(ConnectionPid, Msg) ->
     ConnectionPid ! Msg,
     ok.
 
-<<<<<<< HEAD
 is_ff_enabled() ->
     rabbit_feature_flags:is_enabled(stream_single_active_consumer).
-=======
+
 has_unblock_group_support() ->
     rabbit_feature_flags:is_enabled(stream_sac_coordinator_unblock_group).
->>>>>>> 221f10d2d9 (Unblock group of consumers on super stream partition)

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -971,8 +971,6 @@ open(info, emit_stats,
          StatemData) ->
     Connection1 = emit_stats(Connection, State),
     {keep_state, StatemData#statem_data{connection = Connection1}};
-<<<<<<< HEAD
-=======
 open(info, check_outstanding_requests,
      #statem_data{connection = #stream_connection{outstanding_requests = Requests,
                                                   request_timeout = Timeout} = Connection0} =
@@ -1012,7 +1010,6 @@ open(info, {shutdown, Explanation} = Reason,
                                [self(), Explanation]),
     _ = demonitor_all_streams(Connection),
     {stop, Reason};
->>>>>>> 62d016d3c5 (Introduce timeout for stream server-side requests)
 open(info, Unknown, _StatemData) ->
     rabbit_log_connection:warning("Received unknown message ~p in state ~s",
                                   [Unknown, ?FUNCTION_NAME]),
@@ -2971,33 +2968,20 @@ maybe_send_consumer_update(Transport,
                                            correlation_id_sequence = CorrIdSeq},
                            Consumer,
                            Active,
-<<<<<<< HEAD
-                           true = _Sac,
-                           Extra) ->
-    rabbit_log:debug("SAC subscription ~p, active = ~p",
-                     [SubscriptionId, Active]),
-=======
                            Msg) ->
-<<<<<<< HEAD
     #consumer{configuration = #consumer_configuration{
                                             stream = Stream,
                                             subscription_id = SubscriptionId
                                            }} = Consumer,
     rabbit_log:debug("SAC subscription ~tp on ~tp, active = ~tp, " ++
                      "sending consumer update with correlation ID ~tp",
                      [SubscriptionId, Stream, Active, CorrIdSeq]),
->>>>>>> 221f10d2d9 (Unblock group of consumers on super stream partition)
-    Frame =
-        rabbit_stream_core:frame({request, CorrIdSeq,
-                                  {consumer_update, SubscriptionId, Active}}),
-=======
     #consumer{configuration =
               #consumer_configuration{subscription_id = SubscriptionId}} = Consumer,
     Frame = rabbit_stream_core:frame({request, CorrIdSeq,
                                       {consumer_update, SubscriptionId, Active}}),
 
     Connection1 = register_request(Connection, Msg),
->>>>>>> 62d016d3c5 (Introduce timeout for stream server-side requests)
 
     send(Transport, S, Frame),
     Connection1.

deps/rabbitmq_stream/test/rabbit_stream_SUITE.erl

@@ -131,20 +131,6 @@ end_per_group(_, Config) ->
     rabbit_ct_helpers:run_steps(Config,
                                 rabbit_ct_broker_helpers:teardown_steps()).
 
-<<<<<<< HEAD
-init_per_testcase(_TestCase, Config) ->
-    Config.
-
-end_per_testcase(sac_ff, Config) ->
-    rabbit_ct_broker_helpers:rpc(Config,
-                                 0,
-                                 rabbit_feature_flags,
-                                 enable,
-                                 [stream_single_active_consumer]),
-    ok;
-end_per_testcase(_Test, _Config) ->
-    ok.
-=======
 init_per_testcase(close_connection_on_consumer_update_timeout = TestCase, Config) ->
     ok = rabbit_ct_broker_helpers:rpc(Config,
                                       0,
@@ -161,10 +147,19 @@ end_per_testcase(close_connection_on_consumer_update_timeout = TestCase, Config)
                                       application,
                                       set_env,
                                       [rabbitmq_stream, request_timeout, 60000]),
+    _ = rabbit_ct_broker_helpers:rpc(Config,
+                                     0,
+                                     rabbit_feature_flags,
+                                     enable,
+                                     [stream_single_active_consumer]),
     rabbit_ct_helpers:testcase_finished(Config, TestCase);
 end_per_testcase(TestCase, Config) ->
+    _ = rabbit_ct_broker_helpers:rpc(Config,
+                                     0,
+                                     rabbit_feature_flags,
+                                     enable,
+                                     [stream_single_active_consumer]),
     rabbit_ct_helpers:testcase_finished(Config, TestCase).
->>>>>>> 62d016d3c5 (Introduce timeout for stream server-side requests)
 
 test_global_counters(Config) ->
     Stream = atom_to_binary(?FUNCTION_NAME, utf8),