Function workflow_default

test/pubsub-disruption/mzcompose.py:77–173

Test that the system is able to make progress in the face of PubSub disruptions.

(c: Composition, parser: WorkflowArgumentParser)

def workflow_default(c: Composition, parser: WorkflowArgumentParser) -> None:
    """Test that the system is able to make progress in the face of PubSub disruptions."""
    parser.add_argument("disruptions", nargs="*", default=[d.name for d in disruptions])

    args = parser.parse_args()

    for disruption in selected_by_name(args.disruptions, disruptions):
        c.down(destroy_volumes=True)
        c.up("redpanda", "materialized", Service("testdrive", idle=True))

        toxiproxy_start(c)

        c.testdrive(input=SCHEMA + dedent("""
                > CREATE TABLE t1 (f1 INTEGER, f2 INTEGER);
                $ kafka-create-topic topic=pubsub-disruption partitions=4

                > CREATE CONNECTION IF NOT EXISTS csr_conn
                  TO CONFLUENT SCHEMA REGISTRY (URL '${testdrive.schema-registry-url}');

                > CREATE CONNECTION IF NOT EXISTS kafka_conn
                  TO KAFKA (BROKER '${testdrive.kafka-addr}', SECURITY PROTOCOL PLAINTEXT);

                > INSERT INTO t1 SELECT generate_series, 1 FROM generate_series(1,1000000);
                $ kafka-ingest format=avro key-format=avro topic=pubsub-disruption schema=${schema} key-schema=${keyschema} start-iteration=1 repeat=1000000
                {"f1": ${kafka-ingest.iteration}} {"f2": 1}

                > CREATE SOURCE s1
                  FROM KAFKA CONNECTION kafka_conn
                  (TOPIC 'testdrive-pubsub-disruption-${testdrive.seed}')

                > CREATE TABLE s1_tbl FROM SOURCE s1 (REFERENCE "testdrive-pubsub-disruption-${testdrive.seed}")
                  FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION csr_conn
                  ENVELOPE UPSERT

                > CREATE MATERIALIZED VIEW v1 AS
                  SELECT COUNT(*) AS c1, COUNT(DISTINCT f1) AS c2, COUNT(DISTINCT f2) AS c3,
                         MIN(f1) AS min1, MIN(f2) AS min2, MAX(f1) AS max1, MAX(f2) AS max2
                  FROM t1;

                > CREATE MATERIALIZED VIEW v2 AS
                  SELECT COUNT(*) AS c1, COUNT(DISTINCT f1) AS c2, COUNT(DISTINCT f2) AS c3,
                         MIN(f1) AS min1, MIN(f2) AS min2, MAX(f1) AS max1, MAX(f2) AS max2
                  FROM s1_tbl;

                > UPDATE t1 SET f2 = 2;
                $ kafka-ingest format=avro key-format=avro topic=pubsub-disruption schema=${schema} key-schema=${keyschema} start-iteration=1 repeat=1000000
                {"f1": ${kafka-ingest.iteration}} {"f2": 2}
                """))

        disruption.breakage(c)

        c.testdrive(input=SCHEMA + dedent("""
                $ set-sql-timeout duration=120s

                > UPDATE t1 SET f2 = 3;
                $ kafka-ingest format=avro key-format=avro topic=pubsub-disruption schema=${schema} key-schema=${keyschema} start-iteration=1 repeat=1000000
                {"f1": ${kafka-ingest.iteration}} {"f2": 3}