Skip to content

Commit b660793

Browse files
maltesandermaltesander
committed
hash node.id offset from rolegroup
1 parent e64d9aa commit b660793

6 files changed

Lines changed: 58 additions & 18 deletions

File tree

‎rust/operator-binary/src/config/command.rs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -11,10 +11,10 @@ use crate::crd::{
1111
STACKABLE_LISTENER_BOOTSTRAP_DIR, STACKABLE_LISTENER_BROKER_DIR, STACKABLE_LOG_DIR,
1212
listener::{KafkaListenerConfig, KafkaListenerName, node_address_cmd},
1313
role::{
14-
KAFKA_ADVERTISED_LISTENERS, KAFKA_BROKER_ID_OFFSET,
15-
KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS, KAFKA_CONTROLLER_QUORUM_VOTERS,
16-
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP, KAFKA_LISTENERS, KAFKA_NODE_ID, KafkaRole,
17-
broker::BROKER_PROPERTIES_FILE, controller::CONTROLLER_PROPERTIES_FILE,
14+
KAFKA_ADVERTISED_LISTENERS, KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS,
15+
KAFKA_CONTROLLER_QUORUM_VOTERS, KAFKA_LISTENER_SECURITY_PROTOCOL_MAP, KAFKA_LISTENERS,
16+
KAFKA_NODE_ID, KAFKA_NODE_ID_OFFSET, KafkaRole, broker::BROKER_PROPERTIES_FILE,
17+
controller::CONTROLLER_PROPERTIES_FILE,
1818
},
1919
security::KafkaTlsSecurity,
2020
v1alpha1,
@@ -87,7 +87,7 @@ fn broker_start_command(
8787
export REPLICA_ID=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$')
8888
cp {config_dir}/{properties_file} /tmp/{properties_file}
8989
90-
echo \"{KAFKA_NODE_ID}=$((REPLICA_ID + {KAFKA_BROKER_ID_OFFSET}))\" >> /tmp/{properties_file}
90+
echo \"{KAFKA_NODE_ID}=$((REPLICA_ID + ${KAFKA_NODE_ID_OFFSET}))\" >> /tmp/{properties_file}
9191
echo \"{KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS}={bootstrap_servers}\" >> /tmp/{properties_file}
9292
echo \"{KAFKA_LISTENERS}={listeners}\" >> /tmp/{properties_file}
9393
echo \"{KAFKA_ADVERTISED_LISTENERS}={advertised_listeners}\" >> /tmp/{properties_file}
@@ -148,7 +148,7 @@ pub fn controller_kafka_container_command(
148148
export REPLICA_ID=$(echo \"$POD_NAME\" | grep -oE '[0-9]+$')
149149
cp {config_dir}/{properties_file} /tmp/{properties_file}
150150
151-
echo \"{KAFKA_NODE_ID}=$REPLICA_ID\" >> /tmp/{properties_file}
151+
echo \"{KAFKA_NODE_ID}=$((REPLICA_ID + ${KAFKA_NODE_ID_OFFSET}))\" >> /tmp/{properties_file}
152152
echo \"{KAFKA_CONTROLLER_QUORUM_BOOTSTRAP_SERVERS}={bootstrap_servers}\" >> /tmp/{properties_file}
153153
echo \"{KAFKA_LISTENERS}={listeners}\" >> /tmp/{properties_file}
154154
echo \"{KAFKA_LISTENER_SECURITY_PROTOCOL_MAP}={listener_security_protocol_map}\" >> /tmp/{properties_file}

‎rust/operator-binary/src/config/mod.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,3 @@
11
pub mod command;
22
pub mod jvm;
3+
pub mod node_id_hasher;

‎rust/operator-binary/src/config/node_id_hasher.rs‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
pub fn node_id_hash32_offset(rolegroup: &str) -> u32 {
2+
let hash = fnv_hash32(rolegroup);
3+
let range = hash & 0x0000FFFF;
4+
let offset = range * 0x0000FFFF;
5+
offset
6+
}
7+
8+
/// Simple FNV-1a hash impl
9+
fn fnv_hash32(input: &str) -> u32 {
10+
const FNV_OFFSET: u32 = 0x811c9dc5;
11+
const FNV_PRIME: u32 = 0x01000193;
12+
13+
let mut hash = FNV_OFFSET;
14+
for byte in input.as_bytes() {
15+
hash ^= u32::from(*byte);
16+
hash = hash.wrapping_mul(FNV_PRIME);
17+
}
18+
hash
19+
}

‎rust/operator-binary/src/crd/mod.rs‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,13 @@ use stackable_operator::{
2424
versioned::versioned,
2525
};
2626

27-
use crate::crd::{
28-
authorization::KafkaAuthorization,
29-
role::{KafkaRole, broker::BrokerConfigFragment, controller::ControllerConfigFragment},
30-
tls::KafkaTls,
27+
use crate::{
28+
config::node_id_hasher::node_id_hash32_offset,
29+
crd::{
30+
authorization::KafkaAuthorization,
31+
role::{KafkaRole, broker::BrokerConfigFragment, controller::ControllerConfigFragment},
32+
tls::KafkaTls,
33+
},
3134
};
3235

3336
pub const DOCKER_IMAGE_BASE_NAME: &str = "kafka";
@@ -255,6 +258,7 @@ impl v1alpha1::KafkaCluster {
255258
role_group_service_name: rolegroup_ref.object_name(),
256259
replica: i,
257260
cluster_domain: cluster_info.cluster_domain.clone(),
261+
node_id: node_id_hash32_offset(rolegroup_name) + u32::from(i),
258262
})
259263
})
260264
.collect(),
@@ -275,6 +279,7 @@ impl v1alpha1::KafkaCluster {
275279
role_group_service_name: rolegroup_ref.object_name(),
276280
replica: i,
277281
cluster_domain: cluster_info.cluster_domain.clone(),
282+
node_id: node_id_hash32_offset(rolegroup_name) + u32::from(i),
278283
})
279284
})
280285
.collect(),
@@ -291,6 +296,7 @@ pub struct KafkaPodDescriptor {
291296
role_group_service_name: String,
292297
replica: u16,
293298
cluster_domain: DomainName,
299+
node_id: u32,
294300
}
295301

296302
impl KafkaPodDescriptor {
@@ -332,16 +338,16 @@ impl KafkaPodDescriptor {
332338
// TODO(@maltesander): Even though the used Uuid states to be type 4 it does not work... 0000000000-00000000000 works...
333339
pub fn as_voter(&self, port: u16) -> String {
334340
format!(
335-
"{replica}@{fqdn}:{port}:0000000000-{replica:0>11}",
336-
replica = self.replica,
341+
"{node_id}@{fqdn}:{port}:0000000000-{node_id:0>11}",
342+
node_id = self.node_id,
337343
fqdn = self.fqdn(),
338344
)
339345
}
340346

341347
pub fn as_quorum_voter(&self, port: u16) -> String {
342348
format!(
343-
"{replica}@{fqdn}:{port}",
344-
replica = self.replica,
349+
"{node_id}@{fqdn}:{port}",
350+
node_id = self.node_id,
345351
fqdn = self.fqdn(),
346352
)
347353
}

‎rust/operator-binary/src/crd/role/mod.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,8 @@ use crate::{
3030
v1alpha1,
3131
};
3232

33-
/// Broker and Kafka node.id properties should not clash; This is an offset for brokers.
34-
pub const KAFKA_BROKER_ID_OFFSET: u16 = 1000;
33+
/// Env var
34+
pub const KAFKA_NODE_ID_OFFSET: &str = "NODE_ID_OFFSET";
3535

3636
// See: https://kafka.apache.org/documentation/#brokerconfigs
3737
/// The node ID associated with the roles this process is playing when process.roles is non-empty.

‎rust/operator-binary/src/resource/statefulset.rs‎

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,14 +43,20 @@ use stackable_operator::{
4343
};
4444

4545
use crate::{
46-
config::command::{broker_kafka_container_commands, controller_kafka_container_command},
46+
config::{
47+
command::{broker_kafka_container_commands, controller_kafka_container_command},
48+
node_id_hasher::node_id_hash32_offset,
49+
},
4750
crd::{
4851
self, APP_NAME, KAFKA_HEAP_OPTS, LISTENER_BOOTSTRAP_VOLUME_NAME,
4952
LISTENER_BROKER_VOLUME_NAME, LOG_DIRS_VOLUME_NAME, METRICS_PORT, METRICS_PORT_NAME,
5053
STACKABLE_CONFIG_DIR, STACKABLE_DATA_DIR, STACKABLE_LISTENER_BOOTSTRAP_DIR,
5154
STACKABLE_LISTENER_BROKER_DIR, STACKABLE_LOG_CONFIG_DIR, STACKABLE_LOG_DIR,
5255
listener::get_kafka_listener_config,
53-
role::{AnyConfig, KafkaRole, broker::BrokerContainer, controller::ControllerContainer},
56+
role::{
57+
AnyConfig, KAFKA_NODE_ID_OFFSET, KafkaRole, broker::BrokerContainer,
58+
controller::ControllerContainer,
59+
},
5460
security::KafkaTlsSecurity,
5561
v1alpha1,
5662
},
@@ -325,6 +331,10 @@ pub fn build_broker_rolegroup_statefulset(
325331
"CONTAINERDEBUG_LOG_DIRECTORY",
326332
format!("{STACKABLE_LOG_DIR}/containerdebug"),
327333
)
334+
.add_env_var(
335+
KAFKA_NODE_ID_OFFSET,
336+
node_id_hash32_offset(&rolegroup_ref.role_group).to_string(),
337+
)
328338
.add_env_vars(env)
329339
.add_container_ports(container_ports(kafka_security))
330340
.add_volume_mount(LOG_DIRS_VOLUME_NAME, STACKABLE_DATA_DIR)
@@ -663,6 +673,10 @@ pub fn build_controller_rolegroup_statefulset(
663673
"CONTAINERDEBUG_LOG_DIRECTORY",
664674
format!("{STACKABLE_LOG_DIR}/containerdebug"),
665675
)
676+
.add_env_var(
677+
KAFKA_NODE_ID_OFFSET,
678+
node_id_hash32_offset(&rolegroup_ref.role_group).to_string(),
679+
)
666680
.add_env_vars(env)
667681
.add_container_ports(container_ports(kafka_security))
668682
.add_volume_mount(LOG_DIRS_VOLUME_NAME, STACKABLE_DATA_DIR)

0 commit comments

Comments
 (0)