Add failing appid reconfigure test

This test is very flaky, but it is able (with a sufficient number of
runs) to expose a problem on rollover after recovery, after purging and
deleting an appid.

Signed-off-by: Patrick M. Niedzielski <[email protected]>

src/integration-tests/test_reconfigure_domains.py

@@ -44,10 +44,14 @@ def setup_cluster(self, cluster: Cluster):
         self.writer = proxy.create_client("writers")
         self.writer.open(URI_PRIORITY_1, flags=["write,ack"], succeed=True)
         self.writer.open(URI_PRIORITY_2, flags=["write,ack"], succeed=True)
+        self.writer.open(tc.URI_FANOUT_SC, flags=["write,ack"], succeed=True)
 
         self.reader = proxy.create_client("readers")
         self.reader.open(URI_PRIORITY_1, flags=["read"], succeed=True)
         self.reader.open(URI_PRIORITY_2, flags=["read"], succeed=True)
+        self.reader.open(tc.URI_FANOUT_SC_FOO, flags=["read"], succeed=True)
+        self.reader.open(tc.URI_FANOUT_SC_BAR, flags=["read"], succeed=True)
+        self.reader.open(tc.URI_FANOUT_SC_BAZ, flags=["read"], succeed=True)
 
     # Instruct 'writer to 'POST 'n' messages to the domain.
     # Returns 'True' if all of them succeed, and a false-y value otherwise.
@@ -58,6 +62,22 @@ def post_n_msgs(self, uri, n):
         )
         return all(res == Client.e_SUCCESS for res in results)
 
+    # Instruct 'writer' to 'POST' enough messages to the domain to initiate a
+    # rollover, assuming 'data_file_size' is the size of the data file of the
+    # partition that the domain lives on.
+    # Returns 'True' if all of the messages that were sent to the broker
+    # succeed, and a false-y value otherwise.  Note: 'True' does not mean that
+    # the broker truly initiated rollover.
+    def cause_rollover(self, uri, data_file_size):
+        num_msgs = int(data_file_size / 1024) + 1
+        payload = "Q" * 1024
+
+        results = (
+            self.writer.post(uri, payload=[payload], wait_ack=True)
+            for i in range(0, num_msgs)
+        )
+        return all(res == Client.e_SUCCESS for res in results)
+
     # Helper method which tells 'leader' to reload the domain config.
     def reconfigure_to_n_msgs(self, cluster: Cluster, num_msgs, leader_only=True):
         cluster.config.domains[
@@ -460,3 +480,67 @@ def run_consumers(max_attempts: int) -> int:
         # After: queue is []
         num_confirmed = run_consumers(max_attempts=6)
         assert 4 == num_confirmed
+
+    @tweak.cluster.partition_config.max_data_file_size(32 * 1024)
+    def test_reconfigure_fanout_appids(self, multi_node: Cluster):
+        leader = multi_node.last_known_leader
+
+        admin = AdminClient()
+        admin.connect(*multi_node.admin_endpoint)
+
+        assert self.post_n_msgs(tc.URI_FANOUT_SC, 16)
+
+        self.reader.confirm(tc.URI_FANOUT_SC_FOO, "+16", succeed=True)
+        self.reader.confirm(tc.URI_FANOUT_SC_BAR, "+16", succeed=True)
+        self.reader.confirm(tc.URI_FANOUT_SC_BAZ, "+16", succeed=True)
+
+        assert self.post_n_msgs(tc.URI_FANOUT_SC, 16)
+
+        res = admin.send_admin(
+            f"DOMAINS DOMAIN {tc.DOMAIN_FANOUT_SC}"
+            f" QUEUE {tc.TEST_QUEUE}"
+            f" PURGE {tc.TEST_APPIDS[1]}"
+        )
+        assert "Purged 16 message(s)" in res
+
+        # Purge and delete appid
+        multi_node.config.domains[
+            tc.DOMAIN_FANOUT_SC
+        ].definition.parameters.mode.fanout.app_ids = [tc.TEST_APPIDS[0]]
+        multi_node.reconfigure_domain(
+            tc.DOMAIN_FANOUT_SC, leader_only=False, succeed=True
+        )
+
+        res = admin.send_admin(f"DOMAINS DOMAIN {tc.DOMAIN_FANOUT_SC} INFOs")
+        assert tc.TEST_APPIDS[1] not in res
+        assert tc.TEST_APPIDS[2] not in res
+
+        # Simulate replicas crashing and undergoing recovery from the journal
+        # file.
+        nodes = multi_node.nodes()
+        for node in nodes:
+            if node == node.last_known_leader:
+                continue
+            node.check_exit_code = False
+            node.kill()
+
+        time.sleep(5)
+
+        for node in nodes:
+            if node == node.last_known_leader:
+                continue
+            node.start()
+            node.check_exit_code = True
+
+        # Rollover
+        self.cause_rollover(
+            tc.URI_FANOUT_SC,
+            multi_node.config.definition.partition_config.max_data_file_size,
+        )
+
+        rollover_log = leader.capture("rollover")
+        assert rollover_log is not None
+
+        self.reader.confirm(tc.URI_FANOUT_SC_FOO, "*", succeed=True)
+        self.reader.confirm(tc.URI_FANOUT_SC_BAR, "*", succeed=True)
+        self.reader.confirm(tc.URI_FANOUT_SC_BAZ, "*", succeed=True)