zippy: Fix various issues

misc/python/materialize/zippy/backup_and_restore_actions.py

@@ -39,6 +39,7 @@ def run(self, c: Composition, state: State) -> None:
                 system_parameter_defaults=state.system_parameter_defaults,
                 sanity_restart=False,
                 restart="on-failure",
+                additional_system_parameter_defaults=state.additional_system_parameter_defaults,
                 metadata_store="cockroach",
                 default_replication_factor=2,
             )

misc/python/materialize/zippy/balancerd_actions.py

@@ -15,6 +15,20 @@
 from materialize.zippy.mz_capabilities import MzIsRunning
 
 
+def balancerd_service(state: State) -> Balancerd:
+    """Balancerd configured to route to the current Mz service."""
+    return Balancerd(
+        https_resolver_template=f"{state.mz_service}:6876",
+        static_resolver_addr=f"{state.mz_service}:6875",
+    )
+
+
+def restart_balancerd(c: Composition, state: State) -> None:
+    with c.override(balancerd_service(state)):
+        c.kill("balancerd")
+        c.up("balancerd")
+
+
 class BalancerdStart(Action):
     """Starts balancerd"""
 
@@ -27,12 +41,7 @@ def incompatible_with(cls) -> set[type[Capability]]:
         return {BalancerdIsRunning}
 
     def run(self, c: Composition, state: State) -> None:
-        with c.override(
-            Balancerd(
-                https_resolver_template=f"{state.mz_service}:6876",
-                static_resolver_addr=f"{state.mz_service}:6875",
-            )
-        ):
+        with c.override(balancerd_service(state)):
             c.up("balancerd")
 
     def provides(self) -> list[Capability]:
@@ -65,11 +74,4 @@ def requires(cls) -> set[type[Capability]]:
         return {BalancerdIsRunning, MzIsRunning}
 
     def run(self, c: Composition, state: State) -> None:
-        with c.override(
-            Balancerd(
-                https_resolver_template=f"{state.mz_service}:6876",
-                static_resolver_addr=f"{state.mz_service}:6875",
-            )
-        ):
-            c.kill("balancerd")
-            c.up("balancerd")
+        restart_balancerd(c, state)

misc/python/materialize/zippy/debezium_actions.py

@@ -35,20 +35,6 @@ def run(self, c: Composition, state: State) -> None:
         c.up("debezium")
 
 
-class DebeziumStop(Action):
-    """Stop the Debezium instance."""
-
-    @classmethod
-    def requires(cls) -> set[type[Capability]]:
-        return {DebeziumRunning}
-
-    def withholds(self) -> set[type[Capability]]:
-        return {DebeziumRunning}
-
-    def run(self, c: Composition, state: State) -> None:
-        c.kill("debezium")
-
-
 class CreateDebeziumSource(Action):
     """Creates a Debezium source in Materialized."""
 
@@ -63,39 +49,47 @@ def requires(cls) -> set[type[Capability]]:
         }
 
     def __init__(self, capabilities: Capabilities) -> None:
-        # To avoid conflicts, we make sure the postgres table and the debezium source have matching names
-        postgres_table = random.choice(capabilities.get(PostgresTableExists))
-        cluster_name = random.choice(source_capable_clusters(capabilities))
-        debezium_source_name = f"debezium_source_{postgres_table.name}"
-        this_debezium_source = DebeziumSourceExists(name=debezium_source_name)
-
-        existing_debezium_sources = [
-            s
-            for s in capabilities.get(DebeziumSourceExists)
-            if s.name == this_debezium_source.name
-        ]
-
-        if len(existing_debezium_sources) == 0:
-            self.new_debezium_source = True
-
-            self.debezium_source = this_debezium_source
-            self.postgres_table = postgres_table
-            self.debezium_source.postgres_table = self.postgres_table
-            self.cluster_name = cluster_name
-        elif len(existing_debezium_sources) == 1:
-            self.new_debezium_source = False
-
-            self.debezium_source = existing_debezium_sources[0]
-            assert self.debezium_source.postgres_table is not None
-            self.postgres_table = self.debezium_source.postgres_table
-        else:
-            raise RuntimeError("More than one Debezium source exists")
+        # ENVELOPE DEBEZIUM requires a key, so only Postgres tables with a
+        # primary key can back a Debezium source.
+        pk_tables = [t for t in capabilities.get(PostgresTableExists) if t.has_pk]
+
+        self.new_debezium_source = False
+        self.debezium_source: DebeziumSourceExists | None = None
+
+        if pk_tables:
+            # To avoid conflicts, we make sure the postgres table and the debezium source have matching names
+            postgres_table = random.choice(pk_tables)
+            cluster_name = random.choice(source_capable_clusters(capabilities))
+            debezium_source_name = f"debezium_source_{postgres_table.name}"
+            this_debezium_source = DebeziumSourceExists(name=debezium_source_name)
+
+            existing_debezium_sources = [
+                s
+                for s in capabilities.get(DebeziumSourceExists)
+                if s.name == this_debezium_source.name
+            ]
+
+            if len(existing_debezium_sources) == 0:
+                self.new_debezium_source = True
+
+                self.debezium_source = this_debezium_source
+                self.postgres_table = postgres_table
+                self.debezium_source.postgres_table = self.postgres_table
+                self.cluster_name = cluster_name
+            elif len(existing_debezium_sources) == 1:
+                self.debezium_source = existing_debezium_sources[0]
+                assert self.debezium_source.postgres_table is not None
+                self.postgres_table = self.debezium_source.postgres_table
+            else:
+                raise RuntimeError("More than one Debezium source exists")
 
         super().__init__(capabilities)
 
     def run(self, c: Composition, state: State) -> None:
         if self.new_debezium_source:
-            c.testdrive(dedent(f"""
+            assert self.debezium_source is not None
+            c.testdrive(
+                dedent(f"""
                     $ http-request method=POST url=http://debezium:8083/connectors content-type=application/json
                     {{
                       "name": "{self.debezium_source.name}",
@@ -113,7 +107,7 @@ def run(self, c: Composition, state: State) -> None:
                         "publication.name": "dbz_publication_{self.debezium_source.name}",
                         "publication.autocreate.mode": "filtered",
                         "slot.name" : "slot_{self.postgres_table.name}",
-                        "database.history.kafka.bootstrap.servers": "kafka:9092",
+                        "database.history.kafka.bootstrap.servers": "redpanda:9092",
                         "database.history.kafka.topic": "schema-changes.history",
                         "truncate.handling.mode": "include",
                         "decimal.handling.mode": "precise",
@@ -134,7 +128,12 @@ def run(self, c: Composition, state: State) -> None:
                     > CREATE TABLE {self.debezium_source.get_name_for_query()} FROM SOURCE {self.debezium_source.name} (REFERENCE "postgres.public.{self.postgres_table.name}")
                       FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn
                       ENVELOPE DEBEZIUM
-                    """))
+                    """),
+                mz_service=state.mz_service,
+            )
 
     def provides(self) -> list[Capability]:
-        return [self.debezium_source] if self.new_debezium_source else []
+        if self.new_debezium_source:
+            assert self.debezium_source is not None
+            return [self.debezium_source]
+        return []

misc/python/materialize/zippy/framework.py

@@ -7,7 +7,9 @@
 # the Business Source License, use of this software will be governed
 # by the Apache License, Version 2.0.
 
+import os
 import random
+import threading
 from collections.abc import Sequence
 from datetime import datetime, timedelta
 from typing import TYPE_CHECKING, TypeVar
@@ -19,6 +21,20 @@
     from materialize.zippy.scenarios import Scenario
 
 
+def ci_additional_system_parameter_defaults() -> dict[str, str]:
+    """Parse the CI_MZ_SYSTEM_PARAMETER_DEFAULT environment variable."""
+    defaults = {}
+    system_parameter_default = os.getenv("CI_MZ_SYSTEM_PARAMETER_DEFAULT", "")
+    if system_parameter_default:
+        for val in system_parameter_default.split(";"):
+            x = val.split("=", maxsplit=1)
+            assert (
+                len(x) == 2
+            ), f"CI_MZ_SYSTEM_PARAMETER_DEFAULT '{val}' should be the format <key>=<val>"
+            defaults[x[0]] = x[1]
+    return defaults
+
+
 class State:
     mz_service: str
     deploy_generation: int
@@ -28,8 +44,15 @@ def __init__(self):
         self.mz_service = "materialized"
         self.deploy_generation = 0
         self.system_parameter_defaults = get_default_system_parameters()
-        self.iceberg_username: str | None = None
-        self.iceberg_key: str | None = None
+        # Every action that (re)starts Mz must pass these along, otherwise a
+        # feature flag under test silently reverts to its builtin default.
+        self.additional_system_parameter_defaults = (
+            ci_additional_system_parameter_defaults()
+        )
+        # Threads spawned by actions (e.g. parallel Kafka ingestion). They are
+        # joined at the end of the test and any errors they collected fail it.
+        self.background_threads: list[threading.Thread] = []
+        self.background_errors: list[Exception] = []
 
 
 class Capability:
@@ -101,8 +124,10 @@ def get_free_capability_name(
         ]
         existing_object_names = self.get_capability_names(capability)
         remaining_object_names = set(all_object_names) - set(existing_object_names)
+        # sorted() so that the same --seed picks the same names regardless of
+        # set iteration order (which depends on PYTHONHASHSEED).
         return (
-            random.choice(list(remaining_object_names))
+            random.choice(sorted(remaining_object_names))
             if len(remaining_object_names) > 0
             else None
         )
@@ -191,6 +216,10 @@ def __init__(
         self._actions: list[Action] = []
         self._final_actions: list[Action] = []
         self._capabilities = Capabilities()
+        # The exact Capability instances each action added at generation time.
+        # Trimming in run() removes by identity, and provides() may build new
+        # instances on each call, so the originals must be remembered here.
+        self._provided_capabilities: dict[Action, list[Capability]] = {}
         self._actions_with_weight: dict[ActionOrFactory, float] = (
             self._scenario.actions_with_weight()
         )
@@ -215,11 +244,10 @@ def generate_actions(self, action_def: ActionOrFactory) -> list[Action]:
             )
 
         for action in actions:
-            print("test:", action)
-            self._capabilities._extend(action.provides())
-            print(" - ", self._capabilities, action.provides())
+            provided = action.provides()
+            self._provided_capabilities[action] = provided
+            self._capabilities._extend(provided)
             self._capabilities._remove(action.withholds())
-            print(" - ", self._capabilities, action.withholds())
 
         return actions
 
@@ -237,10 +265,19 @@ def run(self, c: Composition) -> None:
                 executed_count = i + 1
                 break
 
+        # Surface failures from background threads instead of dropping them.
+        for thread in self._state.background_threads:
+            thread.join()
+        if self._state.background_errors:
+            raise self._state.background_errors[0]
+
         # Remove capability effects of actions that were never executed,
         # so that finalization only validates objects that actually exist.
+        # Capabilities withheld by unexecuted actions are not restored. That
+        # is fine because only the *IsRunning capabilities are ever withheld
+        # and finalization unconditionally starts those services again.
         for action in self._actions[executed_count:]:
-            for cap in action.provides():
+            for cap in self._provided_capabilities[action]:
                 self._capabilities.remove_capability_instance(cap)
 
         # Generate finalization actions now, after the main loop, so that

misc/python/materialize/zippy/iceberg_actions.py

@@ -45,8 +45,6 @@ def run(self, c: Composition, state: State) -> None:
             Service("polaris", idle=True),
         )
         username, key = setup_polaris_for_iceberg(c)
-        state.iceberg_username = username
-        state.iceberg_key = key
 
         c.testdrive(
             dedent(f"""

misc/python/materialize/zippy/kafka_actions.py

@@ -12,8 +12,6 @@
 import threading
 from textwrap import dedent
 
-import numpy as np
-
 from materialize.mzcompose.composition import Composition
 from materialize.zippy.framework import (
     Action,
@@ -55,20 +53,6 @@ def run(self, c: Composition, state: State) -> None:
         c.up("redpanda")
 
 
-class KafkaStop(Action):
-    """Stop the Kafka instance."""
-
-    @classmethod
-    def requires(cls) -> set[type[Capability]]:
-        return {KafkaRunning}
-
-    def withholds(self) -> set[type[Capability]]:
-        return {KafkaRunning}
-
-    def run(self, c: Composition, state: State) -> None:
-        c.kill("redpanda")
-
-
 class CreateTopicParameterized(ActionFactory):
     """Creates a Kafka topic and decides on the envelope that will be used."""
 
@@ -139,8 +123,10 @@ def requires(cls) -> set[type[Capability]]:
     def __init__(self, capabilities: Capabilities) -> None:
         self.topic = random.choice(capabilities.get(TopicExists))
         self.delta = random.randint(1, 10000)
-        # This gives 67% pads of up to 10 bytes, 25% of up to 100 bytes and outliers up to 256 bytes
-        self.pad = min(np.random.zipf(1.6, 1)[0], 256) * random.choice(
+        # Heavy-tailed pad sizes: mostly up to 10 bytes, some up to 100 bytes
+        # and outliers up to 256 bytes. int(paretovariate(0.6)) approximates a
+        # zipf(1.6) sample while staying reproducible under random.seed().
+        self.pad = min(int(random.paretovariate(0.6)), 256) * random.choice(
             string.ascii_letters
         )
         super().__init__(capabilities)
@@ -167,7 +153,17 @@ def run(self, c: Composition, state: State) -> None:
             """)
 
         if self.parallel():
-            threading.Thread(target=c.testdrive, args=[testdrive_str]).start()
+            mz_service = state.mz_service
+
+            def ingest() -> None:
+                try:
+                    c.testdrive(testdrive_str, mz_service=mz_service)
+                except Exception as e:
+                    state.background_errors.append(e)
+
+            thread = threading.Thread(target=ingest)
+            thread.start()
+            state.background_threads.append(thread)
         else:
             c.testdrive(testdrive_str, mz_service=state.mz_service)
 

misc/python/materialize/zippy/mysql_actions.py

@@ -40,20 +40,6 @@ def run(self, c: Composition, state: State) -> None:
         )
 
 
-class MySqlStop(Action):
-    """Stop the MySQL instance."""
-
-    @classmethod
-    def requires(cls) -> set[type[Capability]]:
-        return {MySqlRunning}
-
-    def withholds(self) -> set[type[Capability]]:
-        return {MySqlRunning}
-
-    def run(self, c: Composition, state: State) -> None:
-        c.kill("mysql")
-
-
 class MySqlRestart(Action):
     """Restart the MySql instance."""
 
@@ -80,8 +66,7 @@ def __init__(self, capabilities: Capabilities) -> None:
 
         if len(existing_mysql_tables) == 0:
             self.new_mysql_table = True
-            # A PK is now required for Debezium
-            this_mysql_table.has_pk = True
+            this_mysql_table.has_pk = random.choice([True, False])
 
             self.mysql_table = this_mysql_table
         elif len(existing_mysql_tables) == 1: