Kafka

Thanks for the explanation. Slight digression and perhaps a silly question w.r.t. consumers. Since multiple groups are possible, at a high level, the broker effectively sends data of a given topic-partition to multiple consumers while keeping track of offsets. So, why not let consumers specify the partition ID they want to consume? Is the concept of consumer groups only because the consumers wouldn't know in advance how many partitions exist in a topic? Best regards. On Wed, Jan 22, 2025 at 7:41 AM Greg Harris <greg.harris@aiven.io.invalid> wrote: > Hi, > > Thanks for the follow up. > > By "classic" I meant the protocol implemented by the SyncGroup/JoinGroup > API [1]. It's a general group protocol that is still fully supported in > Kraft, and at this time has no intention of being deprecated. > It's called "classic" to distinguish it from the newer KIP-848 [2] > ConsumerGroupHeartbeat API and ...

Hi, Thanks for the follow up. By "classic" I meant the protocol implemented by the SyncGroup/JoinGroup API [1]. It's a general group protocol that is still fully supported in Kraft, and at this time has no intention of being deprecated. It's called "classic" to distinguish it from the newer KIP-848 [2] ConsumerGroupHeartbeat API and the share group protocol used in KIP-932 [3]. It is my understanding that these other protocols are _not_ a synchronizing barrier in the same way the "classic" protocol is. Hope this helps, Greg [1] https://github.com/apache/kafka/blob/adb033211497e539725366960e1013a4638de59f/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java [2] https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol [3] https://cwiki.apache.org/confluence/display/KAFKA/KIP-932%3A+Queues+for+Kafka On Tue, Jan 21, 2025 at 4...

Thanks. By "classic" you mean pre-KRaft consensus? What is the "current" Kafka Group protocol? Best regards. On Tue, Jan 21, 2025 at 9:55 PM Greg Harris <greg.harris@aiven.io.invalid> wrote: > Hi, > > Yes you are correct. The "classic" Kafka Group Protocol is a synchronizing > barrier for all members. > All JoinGroup member responses are returned after all JoinGroup member > requests are received. > > Thanks, > Greg > > On Tue, Jan 21, 2025 at 7:10 AM Chain Head < mrchainhead@gmail.com > wrote: > > > Assume that three consumers of a certain group want to connect to a > broker > > for a topic with 3 partitions. After the FindCoordinator API is done, the > > consumers send JoinGroup. Since the broker cannot know in advance how > many > > consumers are expected to join, it waits > group.initial.rebalance.delay.ms > > before starting a rebal...

Hi, Yes you are correct. The "classic" Kafka Group Protocol is a synchronizing barrier for all members. All JoinGroup member responses are returned after all JoinGroup member requests are received. Thanks, Greg On Tue, Jan 21, 2025 at 7:10 AM Chain Head < mrchainhead@gmail.com > wrote: > Assume that three consumers of a certain group want to connect to a broker > for a topic with 3 partitions. After the FindCoordinator API is done, the > consumers send JoinGroup. Since the broker cannot know in advance how many > consumers are expected to join, it waits group.initial.rebalance.delay.ms > before starting a rebalance. > > Therefore, does this mean the JoinGroup API response of each request is > "held" until the waiting period is over? > > Best regards. >

Assume that three consumers of a certain group want to connect to a broker for a topic with 3 partitions. After the FindCoordinator API is done, the consumers send JoinGroup. Since the broker cannot know in advance how many consumers are expected to join, it waits group.initial.rebalance.delay.ms before starting a rebalance. Therefore, does this mean the JoinGroup API response of each request is "held" until the waiting period is over? Best regards.

Apologies for disappearing. This is how I got it working. In hindsight, I should have known better and tried this first. byte[] keyBytes = new byte[record.keySize()]; record.key().get(keyBytes); byte[] valueBytes = new byte[record.valueSize()]; record.value().get(valueBytes); On Sun, Dec 29, 2024 at 7:01 AM David Finnie < david.finnie@infrasoft.com.au > wrote: > Hi Chain Head, > > Are you seeing any errors, or just getting empty strings for k and v? > > If you are seeing empty strings, it could well be that the ByteBuffer > returned by Record.key() and Record.value() have their position set to > the ByteBuffer's length. i.e. at the end of the ByteBuffer. > > Have you checked the value of position() coming back from those > ByteBuffers, and also have you tried creating the string from position 0 > rather than using the position() result? > > (Disclaimer, I haven't tried writing any code to test the values bei...

I want to use Helidon SE 4.1.6 and produce the data to a specific partition of Apache Kafka using the producer. Detail : I have gone through the https://helidon.io/docs/latest/se/reactive-messaging#_kafka_connectorpage and written below code which is able to send the data using kafka producer to the topic. *Now I want to send the data to specific partition of kafka using Helidon SE 4.1.6.* Below code works and can produce data to a topic (No partitions): public class KafkaProducer { public static void init() { String kafkaServer ="localhost:9092"; String topic = "topic2"; Channel<String> toKafka = Channel.<String>builder() .subscriberConfig(KafkaConnector.configBuilder() .bootstrapServers(kafkaServer) .topic(topic) .keySerializer(StringSerializer.class) .valueSerializer(StringSerializ...