ryerraguntla commented on code in PR #4250: URL: https://github.com/apache/iggy/pull/4250#discussion_r4088497528
########## gateways/kafka/src/protocol/handlers/init_producer_id.rs: ########## @@ -0,0 +1,205 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! `InitProducerId` (API key 22). +//! +//! A Java producer sets `enable.idempotence=true` without being asked (KIP-679, default since +//! Kafka 3.0) and sends this before its first record, so answering it is what lets a stock +//! producer start against this gateway at all. The id is handed out and then ignored: delivery +//! stays at-least-once, and no retry is deduplicated. See `docs/IDEMPOTENCE.md`. + +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use bytes::Bytes; +use kafka_protocol::messages::{InitProducerIdRequest, InitProducerIdResponse, ProducerId}; + +use crate::error::Result; +use crate::protocol::api::{ + API_KEY_INIT_PRODUCER_ID, ApiVersionRange, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR, + ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome, +}; +use crate::protocol::bounds_guard::validate_init_producer_id_shape; +use crate::protocol::handlers::{ + decode_guarded, encode_message, handle_versioned_request, is_transactional, +}; + +pub const RANGE: ApiVersionRange = ApiVersionRange { + api_key: API_KEY_INIT_PRODUCER_ID, + min_version: 0, + max_version: 5, +}; + +/// Width of the per-instance counter. The remaining 16 bits of the non-negative range carry the +/// instance number, and bit 63 stays clear because `producer_id` is an `i64` whose `-1` means +/// "no producer id". +const COUNTER_BITS: u32 = 47; +const MAX_COUNTER: u64 = (1 << COUNTER_BITS) - 1; + +/// The epoch every allocated id carries. Epochs only advance when a producer is fenced, which +/// needs the transactional state this gateway does not keep. +const PRODUCER_EPOCH: i16 = 0; + +/// Hands out producer ids that are unique across gateway instances sharing one Iggy cluster. +/// +/// `instance_id` is configured (`IGGY_KAFKA_INSTANCE_ID`), not drawn at startup: a random 16-bit +/// value collides at even odds around 300 instances. +/// +/// The counter starts at the wall clock in milliseconds, not at 0. Kafka keys a producer on +/// `(producer_id, producer_epoch)` and the epoch is always 0 here, so a counter restarting at 0 +/// would hand a restarted gateway's producers the pairs its previous run gave out. Seeding from +/// the clock keeps every new id above the old ones unless the previous run averaged more than one +/// allocation per millisecond of its uptime, or the clock stepped back across the restart, with +/// nothing persisted. +#[derive(Debug)] +pub struct ProducerIdAllocator { + instance_id: u16, + next_counter: AtomicU64, +} + +impl ProducerIdAllocator { + #[must_use] + pub fn new(instance_id: u16) -> Self { + Self { + instance_id, + next_counter: AtomicU64::new(clock_counter()), Review Comment: clock seed = start millis, not last counter. Epoch always 0. Burst N>(T1−T0) or NTP step-back or map_or(0) pre-epoch replays (pid,0). Test :185 1 alloc + 2ms. Pid unused until persist. Fix: persist high-water or bump epoch before #3535. ########## gateways/kafka/src/server.rs: ########## @@ -77,6 +82,7 @@ impl Default for GatewayConfig { read_timeout: Duration::from_secs(15), write_timeout: Duration::from_secs(10), shutdown_drain_timeout: Duration::from_secs(25), + instance_id: 0, Review Comment: default instance_id=0. Two gateways same cluster share 47-bit space. Same-ms start → same (pid,0). Log only :211. Fix: unique IGGY_KAFKA_INSTANCE_ID per replica. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
