ryerraguntla commented on code in PR #4250:
URL: https://github.com/apache/iggy/pull/4250#discussion_r4082308060
##########
gateways/kafka/src/protocol/handlers/produce.rs:
##########
Review Comment:
acks=0 → NoResponse before partition_error_code. Txn 35 never hits wire.
Stub writes nothing. #3535 persist-after-decode stores abortable records
silent. **Fix**: refuse/drop transactional_id before any write.
##########
gateways/kafka/src/protocol/handlers/produce.rs:
##########
@@ -142,6 +144,23 @@ pub fn encode_response(version: i16, req: &ProduceRequest)
-> Result<Bytes> {
encode_message(&resp, version, 512)
}
+/// A transactional batch must never be answered as if it were ordinary
records: nothing here
+/// tracks a last stable offset or writes an abort marker, so an aborted
transaction's records
+/// would reach every consumer. 35 is fatal for the producer; 42 and 43 are
only abortable.
+///
+/// This reads the request-level `transactional_id` only. A record batch
carries its own
+/// transactional bit in attributes, and the records stay opaque bytes here,
so a hand-built
+/// frame setting the bit without the request field still gets the retriable
stub error. Java
+/// and librdkafka both set the request field whenever they set the batch bit,
so no real client
+/// reaches that gap; it has to close before records are ever persisted.
+fn partition_error_code(req: &ProduceRequest) -> i16 {
Review Comment:
guard reads request transactional_id only. Batch IS_TRANSACTIONAL opaque →
6. Java/librdkafka set both; hand-built frame does not. Same #3535 hole. Do not
parse batches on stub hot path.
##########
gateways/kafka/src/protocol/handlers/init_producer_id.rs:
##########
@@ -0,0 +1,173 @@
+// 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 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.
+#[derive(Debug)]
+pub struct ProducerIdAllocator {
+ instance_id: u16,
+ next_counter: AtomicU64,
+}
+
+impl ProducerIdAllocator {
+ #[must_use]
+ pub const fn new(instance_id: u16) -> Self {
+ Self {
+ instance_id,
+ next_counter: AtomicU64::new(0),
Review Comment:
restart AtomicU64::new(0) + epoch 0 replays (pid,epoch). Kafka uniqueness
is that pair. Allocate-only, pid unused. Pool/persist blocker. **Fix**: persist
counter or bump epoch before Produce persist.
--
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]
