feat: Add support for Kafka clusters sharding #1454

olksdr · 2022-09-06T15:26:11Z

This allows to configured many different Kafka clusters per topic. The
basic idea is to provide the numebr of shards and the lower bound for
the range with the kafka config name and the topic name on that cluster

 metrics:
     shards: 65000
     mapping:
         0:
             name: "ingest-metrics-1"
             config: "metric_1"
         25000:
             name: "ingest-metrics-2"
             config: "metrics_2"
         45000:
             name: "ingest-metrics-3"
             config: "metrics_3"

This will also allow to configure one Kafka clusters and use different
topics names on it.

This allows to configured many different Kafka clusters per topic. The basic idea is to provide the numebr of shards and the lower bound for the range with the kafka config name and the topic name on that cluster ``` metrics: shards: 65000 mapping: 0: name: "ingest-metrics-1" config: "metric_1" 25000: name: "ingest-metrics-2" config: "metrics_2" 45000: name: "ingest-metrics-3" config: "metrics_3" ``` This will also allow to configure one Kafka clusters and use different topics names on it. INGEST-1592

…onfigured

flub · 2022-09-09T09:22:09Z

relay-config/src/config.rs

+pub struct Sharded {
+    /// The number of shards used for this topic.
+    shards: u64,
+    /// The Kafka configuration assigned to the specific shard range.


could you describe that the u64 is the start index and the next u64 describes the range. Explicitly calling out what's inclusive (i'm assuming the start u64 is inclusive, the end u64 is excluded and part of the next range). Maybe even write an example out on the struct doc comment like you did in the PR description.

@flub, please, have a look into f967933 if this is something you had in mind?

flub · 2022-09-09T11:33:55Z

relay-server/src/actors/outcome.rs

+                let org_id_hash = hasher.finish();
+                let shard = org_id_hash % shards;
+
+                // should be ok to unwrap since we MUST have at least one range defined


This is reasonable but because it is so far apart from each other in the code I'm a bit uncomfortable about it. One way would be make a custom type for the producers, then it is enforced in that type and relying on that invariant would be a lot nicer and keeps the code interacting with the invariant close together. I guess this would be a newtype over the BTreeMap with a few small methods.

That's a very good point. I will look into it.
It might also simplify the code and removes some duplication.

So, I moved the same functionality under the Sharded producer in store actore d4a05c2.
I do not know if it's a good idea to create a new type over BTreeMap, which can be a bit more complicated to understand.
But I might have to pick your brain on this , and how to implement better.

flub · 2022-09-09T11:44:14Z

relay-server/src/actors/store.rs

+                let mut hasher = FnvHasher::default();
+                std::hash::Hash::hash(&organization_id, &mut hasher);
+                let org_id_hash = hasher.finish();
+                let shard = org_id_hash % shards;


this code exists in two places already, given the nature of it - i suspect both would need updating together all the time - it might be better to create a method somewhere. Maybe also on the newtype around the BTreeMap I already suggested.

moved this into the impl for Sharded producer in d4a05c2

…harded store producer

tests/integration/fixtures/relay.py

relay-server/src/actors/store.rs

jjbayer · 2022-09-12T11:18:03Z

relay-config/src/config.rs

+        /// The maximum number of logical shards for this set of configs.
+        shards: u64,
+        /// The list of the sharded Kafka configs.
+        configs: BTreeMap<u64, KafkaParams<'a>>,


We have this mapping in three different places now, I wonder if it would be possible to have it only once.

@jjbayer I've tried to remove one of the levels of indirection to eliminate of of the repeating BTreeMap, see 0db194e
Please, have another look if this is somewhat better approach.

Yes, I think it makes sense this way.

Now Kafka config with topic name and parameters is produced from TopicAssignment enum

jjbayer · 2022-09-13T11:45:33Z

relay-config/src/config.rs

-        assert!(matches!(
-            kafka_config_profiles,
-            KafkaConfigName::Single { .. }
-        ));


Any way we could keep this assertion?

done in 1c3d5e0.

jjbayer · 2022-09-13T11:46:16Z

relay-config/src/config.rs

-                topic_name: topic.as_str(),
-                kafka_config_name: None,
-            },
+    fn kafka_config_name<'a>(


Should we update the doc comment and the function name, now that this function returns a full config rather than a name?

Good point, done in 1c3d5e0

jjbayer · 2022-09-13T11:47:21Z

relay-config/src/config.rs

+        /// The maximum number of logical shards for this set of configs.
+        shards: u64,
+        /// The list of the sharded Kafka configs.
+        configs: BTreeMap<u64, KafkaParams<'a>>,


Yes, I think it makes sense this way.

jjbayer · 2022-09-13T11:57:57Z

relay-server/src/actors/store.rs

+            Self::Single {
+                topic_name,
+                producer,
+            } => (topic_name.as_str(), Arc::clone(producer)),


nit: You can probably change the signature of get_producer to return (&str, &ThreadedProducer) to save on Arc::clones.

done in 1c3d5e0

untitaker · 2022-09-13T12:24:24Z

tests/integration/fixtures/relay.py

@@ -133,7 +134,7 @@ def inner(
                default_opts.setdefault(key, {}).update(options[key])

        dir = config_dir("relay")
-        dir.join("config.yml").write(json.dumps(default_opts))
+        dir.join("config.yml").write(yaml.dump(default_opts))


we don't have to change this now, but I believe that at some point the configuration will have to become json-serializable. then we will have to revisit the schema. but that's fine

* master: feat: Add support for Kafka clusters sharding (#1454) feat(replays): Fix typo in payload deserializer (#1467)

As a folllowup for #1454 we decided that we do not have to hash the org id to get the sahrd number and simple modulo #shard will be sufficient. As part of this small refactoring the `unwrap` was also removed and Result is returned instead, to propagate the error to the caller if it happens.

As a followup for #1454 we decided that we do not have to hash the org id to get the shard number and simple modulo #shard will be sufficient. As part of this small refactoring the `unwrap` was also removed and Result is returned instead, to propagate the error to the caller if it happens.

olksdr self-assigned this Sep 6, 2022

olksdr added 10 commits September 7, 2022 09:20

ref: Propagate number of shards, remove some duplucation

771de1b

ref: Enable hashing and thread topic name though the kafka config

1e6ee4b

Merge branch 'master' into feat/kafka-sharding-config

baf73cd

Merge branch 'master' into feat/kafka-sharding-config

7e49aa8

ref: Add sharded producers to outcome actor

3934c69

feat: Add changelog entry

9809809

feat: use either org_id or project_id for sharding in outcomes when c…

a1775a7

…onfigured

test: Add integration tests for checking the kafka sharding

0589017

feat: fail to start if the sharding configuration is invalid

c1fb97e

Merge branch 'master' into feat/kafka-sharding-config

191c768

olksdr marked this pull request as ready for review September 9, 2022 06:19

olksdr requested a review from a team September 9, 2022 06:19

flub approved these changes Sep 9, 2022

View reviewed changes

olksdr added 3 commits September 12, 2022 10:03

Merge branch 'master' into feat/kafka-sharding-config

a0d0987

feat: Add better docs for the new config foramt

f967933

ref: Remove some code duplication, and use same functionality under S…

d4a05c2

…harded store producer

jjbayer reviewed Sep 12, 2022

View reviewed changes

ref: Address the review comments: remove code duplication

f0e792c

iker-barriocanal approved these changes Sep 12, 2022

View reviewed changes

ref: Remove one level of inderections

0db194e

Now Kafka config with topic name and parameters is produced from TopicAssignment enum

olksdr requested review from jjbayer and a team September 13, 2022 11:19

Merge branch 'master' into feat/kafka-sharding-config

6f499da

jjbayer approved these changes Sep 13, 2022

View reviewed changes

untitaker reviewed Sep 13, 2022

View reviewed changes

untitaker approved these changes Sep 13, 2022

View reviewed changes

ref: Address review comments

1c3d5e0

olksdr merged commit 6802bd1 into master Sep 14, 2022

olksdr deleted the feat/kafka-sharding-config branch September 14, 2022 05:27

jan-auer added a commit that referenced this pull request Sep 14, 2022

Merge branch 'master' into feat/protocol-transaction-changes

aaf59da

* master: feat: Add support for Kafka clusters sharding (#1454) feat(replays): Fix typo in payload deserializer (#1467)

olksdr mentioned this pull request Sep 15, 2022

ref: Use simple modulo #shards to get the shard number #1476

Merged

feat: Add support for Kafka clusters sharding #1454

feat: Add support for Kafka clusters sharding #1454

Uh oh!

Conversation

olksdr commented Sep 6, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

olksdr Sep 13, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

olksdr Sep 13, 2022 •

edited

Loading