enable source replication for kafka sources #30003

petrosagg · 2024-10-15T14:13:10Z

Motivation

Tips for reviewer

Checklist

This PR has adequate test coverage / QA involvement has been duly considered. (trigger-ci for additional test/nightly runs)
This PR has an associated up-to-date design doc, is a design doc (template), or is sufficiently small to not require a design.
If this PR evolves an existing $T ⇔ Proto$T mapping (possibly in a backwards-incompatible way), then it is tagged with a T-proto label.
If this PR will require changes to cloud orchestration or tests, there is a companion cloud PR to account for those changes that is tagged with the release-blocker label (example).
If this PR includes major user-facing behavior changes, I have pinged the relevant PM to schedule a changelog post.

jkosh44

Adapter parts LGTM

teskje

I looked at the controller and storage changes and they look good, as far as I understand them (and they are surprisingly small!). The only thing that seems suspicious to me is the handling of DroppedIds.

teskje · 2024-11-15T10:06:43Z

src/storage-controller/src/instance.rs

-        assert!(
-            !objects_installed || self.replicas.is_empty(),
-            "replication not supported for storage objects",
-        );


Can you also remove the stale # Panics docs above?

teskje · 2024-11-15T10:07:09Z

src/storage-controller/src/instance.rs

-        assert!(
-            !command.installs_objects() || self.replicas.len() <= 1,
-            "replication not supported for storage objects"
-        );


Can you also remove the stale # Panics docs above?

teskje · 2024-11-15T10:14:25Z

src/storage-controller/src/lib.rs

-                        soft_panic_or_log!(
-                            "DroppedIds for ID {id} but we have neither ingestion nor export \
-                             under that ID"
-                        );


Would be good to have a comment here explaining when this can happen and why it's fine to ignore it.

I assume in a multi-replica cluster all replicas send a DroppedIds and after the first one we have removed the self.collections entry so we end up here? I don't quite understand what the intention behind DroppedIds is though: I thought it was so the controller would know to keep around resources that are still used by replicas, but with this change it would do so only for the fastest replica.

After what we talked about today I added some code with the simplest way I would think of that handles multiple DroppedIds responses.

teskje · 2024-11-15T10:19:32Z

src/storage/src/source/mysql/replication.rs

-        _ => REPLICATION_SERVER_ID_OFFSET,
-    };
+    let mut rng = rand::thread_rng();
+    let server_id: u32 = rng.gen();


The comment above says:

The value does not actually matter since it's irrelevant for GTID-based replication and won't
cause errors if it happens to be the same as another replica in the mysql cluster

Based on this I'd think it should be fine if all replicas running a source used the same server ID. Is that not the case?

I thought I had reverted this change. It does matter in the end, in the sense that if two clients use the same server_id at the same time one of them will get an error. Choosing a random server id was my initial remedy for this issue, but afterwards I thought it was hacky enough that I removed MySQL from being replicated in this PR and I will treat it like postgres, i.e a source that gets scheduled only on the most recent replica of a cluster

aljoscha

Should this change be gated by the new feedback UPSERT source being enabled?

petrosagg · 2024-12-03T23:18:13Z

This PR is currently blocked until we resolve https://github.com/MaterializeInc/database-issues/issues/8798 , since running a replicated kafka source with the old upsert implementation can lead to negative accumulations

aljoscha

The changes looks good! I want to add a sys cfg for enabled multi-replica sources (I can do it). And I don't know about the test changes, would probably be good to get @def- to have a look

aljoscha · 2025-01-29T13:22:44Z

src/sql/src/plan/statement/ddl.rs

@@ -2102,11 +2114,7 @@ fn source_sink_cluster_config(
        Some(in_cluster) => scx.catalog.get_cluster(in_cluster.id),
    };

-    if cluster.replica_ids().len() > 1 {


excellent! feels weird that this was in here...

aljoscha · 2025-01-29T13:49:14Z

src/storage/src/source/reclock.rs

@@ -130,7 +130,8 @@ where

        while *self.upper == [IntoTime::minimum()]
            || (PartialOrder::less_equal(&self.source_upper.frontier(), &new_from_upper)
-                && PartialOrder::less_than(&self.upper, &new_into_upper))
+                && PartialOrder::less_than(&self.upper, &new_into_upper)
+                && self.upper.less_equal(&binding_ts))


why was this change needed?

aljoscha · 2025-01-29T13:50:39Z

test/testdrive/source-sink-clusters.td

@@ -9,6 +9,7 @@

 $ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
 ALTER SYSTEM SET unsafe_enable_unorchestrated_cluster_replicas = true
+ALTER SYSTEM SET enable_create_table_from_source = true


why is this change needed?

def-

Test changes lgtm assuming nightly is green. Triggered a run: https://buildkite.com/materialize/nightly/builds/11000
Edit: needs to be rebased for upgrade tests to work.
data ingest fails because of this PR:

materialize.data_ingest.query_error.QueryError: ('cannot create source in cluster with more than one replica', 'CREATE SOURCE "materialize"."public"."mysql_source0"\n                    IN CLUSTER "quickstart"\n                    FROM MYSQL CONNECTION mysql0\n                    ')

def- · 2025-01-29T16:16:53Z

misc/python/materialize/parallel_workload/parallel_workload.py

@@ -75,7 +76,7 @@ def run(
    rng = random.Random(random.randrange(SEED_RANGE))

    print(
-        f"+++ Running with: --seed={seed} --threads={num_threads} --runtime={runtime} --complexity={complexity.value} --scenario={scenario.value} {'--naughty-identifiers ' if naughty_identifiers else ''} (--host={host})"
+        f"+++ Running with: --seed={seed} --threads={num_threads} --runtime={runtime} --complexity={complexity.value} --scenario={scenario.value} {'--naughty-identifiers ' if naughty_identifiers else ''} --replicas={replicas} (--host={host})"


Should we also add a new Parallel Workload run with --replicas=2/4 in ci/nightly/pipeline.template.yml?

Signed-off-by: Petros Angelatos <[email protected]>

morsapaes · 2025-02-18T13:53:39Z

Closing, as this has been subsumed by #31227. Thanks for laying out the foundation, @petrosagg! 🫶

petrosagg force-pushed the active-replication branch from b0369ea to 1ec2a97 Compare October 21, 2024 11:20

petrosagg force-pushed the active-replication branch 2 times, most recently from 12f07ff to 818631d Compare November 6, 2024 19:18

petrosagg force-pushed the active-replication branch 2 times, most recently from ce9937a to 606c2ce Compare November 14, 2024 18:02

petrosagg changed the title ~~WIP: enable source replication for non-pg sources~~ enable source replication for kafka sources Nov 14, 2024

petrosagg marked this pull request as ready for review November 14, 2024 19:19

petrosagg requested review from a team as code owners November 14, 2024 19:19

petrosagg requested a review from jkosh44 November 14, 2024 19:19

jkosh44 reviewed Nov 14, 2024

View reviewed changes

teskje reviewed Nov 15, 2024

View reviewed changes

aljoscha reviewed Nov 15, 2024

View reviewed changes

petrosagg force-pushed the active-replication branch from 606c2ce to e697cfb Compare December 3, 2024 20:59

petrosagg force-pushed the active-replication branch 3 times, most recently from 72f8657 to a47e9a1 Compare January 16, 2025 19:23

aljoscha reviewed Jan 29, 2025

View reviewed changes

aljoscha mentioned this pull request Jan 29, 2025

storage: allow kafka/loadgen sources on multi-replica clusters, gated by dyncfg #31227

Merged

5 tasks

def- approved these changes Jan 29, 2025

View reviewed changes

petrosagg added 2 commits January 30, 2025 12:23

storage: allow non-pg sources to be scheduled on multi-replica clusters

6bc2963

Signed-off-by: Petros Angelatos <[email protected]>

track active copies of each collection

6f66a97

petrosagg force-pushed the active-replication branch from a47e9a1 to 6f66a97 Compare January 30, 2025 10:24

morsapaes closed this Feb 18, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

enable source replication for kafka sources #30003

enable source replication for kafka sources #30003

petrosagg commented Oct 15, 2024

jkosh44 left a comment

teskje left a comment

teskje Nov 15, 2024

teskje Nov 15, 2024

teskje Nov 15, 2024 •

edited

Loading

petrosagg Dec 3, 2024

teskje Nov 15, 2024

petrosagg Nov 22, 2024

aljoscha left a comment

petrosagg commented Dec 3, 2024

aljoscha left a comment

aljoscha Jan 29, 2025

aljoscha Jan 29, 2025

aljoscha Jan 29, 2025

def- left a comment •

edited

Loading

def- Jan 29, 2025

morsapaes commented Feb 18, 2025

enable source replication for kafka sources #30003

enable source replication for kafka sources #30003

Conversation

petrosagg commented Oct 15, 2024

Motivation

Tips for reviewer

Checklist

jkosh44 left a comment

Choose a reason for hiding this comment

teskje left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

teskje Nov 15, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aljoscha left a comment

Choose a reason for hiding this comment

petrosagg commented Dec 3, 2024

aljoscha left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

def- left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

morsapaes commented Feb 18, 2025

teskje Nov 15, 2024 •

edited

Loading

def- left a comment •

edited

Loading