You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org> on 2024/12/16 10:30:20 UTC
[PR] First draft of system tests for the new share consumer client [kafka]
chirag-wadhwa5 opened a new pull request, #18209:
URL: https://github.com/apache/kafka/pull/18209
This is the first draft for System Tests for the new share consumer client introduced as part of KIP-932
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Infrastructure for system tests for the new share consumer client [kafka]
Posted by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org>.
chirag-wadhwa5 commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1918820386
##########
tests/kafkatest/tests/verifiable_share_consumer_test.py:
##########
@@ -0,0 +1,106 @@
+# 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.
+
+from ducktape.utils.util import wait_until
+
+from kafkatest.tests.kafka_test import KafkaTest
+from kafkatest.services.verifiable_producer import VerifiableProducer
+from kafkatest.services.verifiable_share_consumer import VerifiableShareConsumer
+from kafkatest.services.kafka import TopicPartition
+
+class VerifiableShareConsumerTest(KafkaTest):
+ PRODUCER_REQUEST_TIMEOUT_SEC = 30
+
+ def __init__(self, test_context, num_consumers=1, num_producers=0, **kwargs):
+ super(VerifiableShareConsumerTest, self).__init__(test_context, **kwargs)
+ self.num_consumers = num_consumers
+ self.num_producers = num_producers
+
+ def _all_partitions(self, topic, num_partitions):
+ partitions = set()
+ for i in range(num_partitions):
+ partitions.add(TopicPartition(topic=topic, partition=i))
+ return partitions
+
+ def _partitions(self, assignment):
+ partitions = []
+ for parts in assignment.values():
+ partitions += parts
+ return partitions
+
+ def valid_assignment(self, topic, num_partitions, assignment):
+ all_partitions = self._all_partitions(topic, num_partitions)
+ partitions = self._partitions(assignment)
+ return len(partitions) == num_partitions and set(partitions) == all_partitions
+
+ def min_cluster_size(self):
+ """Override this since we're adding services outside of the constructor"""
+ return super(VerifiableShareConsumerTest, self).min_cluster_size() + self.num_consumers + self.num_producers
+
+ def setup_share_group(self, topic, acknowledgement_mode="auto", group_id="test_group_id", offset_reset_strategy="", **kwargs):
+ return VerifiableShareConsumer(self.test_context, self.num_consumers, self.kafka,
+ topic, group_id, acknowledgement_mode=acknowledgement_mode,
+ offset_reset_strategy=offset_reset_strategy, log_level="TRACE", **kwargs)
+
+ def setup_producer(self, topic, max_messages=-1, throughput=500):
+ return VerifiableProducer(self.test_context, self.num_producers, self.kafka, topic,
+ max_messages=max_messages, throughput=throughput,
+ request_timeout_sec=self.PRODUCER_REQUEST_TIMEOUT_SEC,
+ log_level="DEBUG")
+
+ def await_produced_messages(self, producer, min_messages=1000, timeout_sec=10):
+ current_acked = producer.num_acked
+ wait_until(lambda: producer.num_acked >= current_acked + min_messages, timeout_sec=timeout_sec,
+ err_msg="Timeout awaiting messages to be produced and acked")
+
+ def await_consumed_messages(self, consumer, min_messages=1, timeout_sec=10, total=False):
+ current_total = 0
+ if total is False:
+ current_total = consumer.total_consumed()
+ wait_until(lambda: consumer.total_consumed() >= current_total + min_messages,
+ timeout_sec=timeout_sec,
+ err_msg="Timed out waiting for consumption")
+
+ def await_consumed_messages_by_a_consumer(self, consumer, node, min_messages=1, timeout_sec=10, total=False):
Review Comment:
Thanks for the review. If you seethe usage of `await_consumed_messages_by_a_consumer`, it is used in the methods `rolling_bounce_share_consumers` and `bounce_all_share_consumers`. `await_consumed_messages` waits for consumption of messages, that could be from any of the share consumers in the group, while `await_consumed_messages_by_a_consumer` provides a specific node and waits for that particular consume to consume some messages. So, in the tests where share consumers are bounced (restarted), this gives us a guarantee that the consumer has started successfully and is consuming messages after the restart.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Infrastructure for system tests for the new share consumer client [kafka]
Posted by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org>.
chirag-wadhwa5 commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1919795494
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH
+
+class ConsumerState:
+ Started = 1
+ Dead = 2
+
+class ShareConsumerEventHandler(object):
+
+ def __init__(self, node, idx, state=ConsumerState.Dead):
+ self.node = node
+ self.idx = idx
+ self.total_consumed = 0
+ self.total_acknowledged = 0
+ self.total_acknowledged_failed = 0
+ self.consumed_per_partition = {}
+ self.acknowledged_per_partition = {}
+ self.acknowledged_per_partition_failed = {}
+ self.state = state
+
+ def handle_shutdown_complete(self, node=None, logger=None):
+ self.state = ConsumerState.Dead
+ if node is not None and logger is not None:
+ logger.debug("Shut down %s" % node.account.hostname)
+
+ def handle_startup_complete(self, node, logger):
+ self.state = ConsumerState.Started
+ logger.debug("Started %s" % node.account.hostname)
+
+ def handle_offsets_acknowledged(self, event, node, logger):
+ if event["success"]:
+ self.total_acknowledged += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition[topic_partition] = self.acknowledged_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ else:
+ self.total_acknowledged_failed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition_failed[topic_partition] = self.acknowledged_per_partition_failed.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ logger.debug("Offset acknowledgement failed for: %s" % (node.account.hostname))
+
+ def handle_records_consumed(self, event, node, logger):
+ self.total_consumed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.consumed_per_partition[topic_partition] = self.consumed_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets consumed for %s" % (node.account.hostname))
+
+
+ def handle_kill_process(self, clean_shutdown):
+ # if the shutdown was clean, then we expect the explicit
+ # shutdown event from the consumer
+ if not clean_shutdown:
+ self.handle_shutdown_complete()
+
+class VerifiableShareConsumer(KafkaPathResolverMixin, VerifiableClientMixin, BackgroundThreadService):
+ """This service wraps org.apache.kafka.tools.VerifiableShareConsumer for use in
+ system testing.
+
+ NOTE: this class should be treated as a PUBLIC API. Downstream users use
+ this service both directly and through class extension, so care must be
+ taken to ensure compatibility.
+ """
+
+ PERSISTENT_ROOT = "/mnt/verifiable_share_consumer"
+ STDOUT_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stdout")
+ STDERR_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stderr")
+ LOG_DIR = os.path.join(PERSISTENT_ROOT, "logs")
+ LOG_FILE = os.path.join(LOG_DIR, "verifiable_share_consumer.log")
+ LOG4J_CONFIG = os.path.join(PERSISTENT_ROOT, "tools-log4j.properties")
+ CONFIG_FILE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.properties")
+
+ logs = {
+ "verifiable_share_consumer_stdout": {
+ "path": STDOUT_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_stderr": {
+ "path": STDERR_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_log": {
+ "path": LOG_FILE,
+ "collect_default": True}
+ }
+
+ def __init__(self, context, num_nodes, kafka, topic, group_id,
+ max_messages=-1, acknowledgement_mode="auto", offset_reset_strategy="",
+ version=DEV_BRANCH, stop_timeout_sec=60, log_level="INFO", jaas_override_variables=None,
+ on_record_consumed=None):
+ """
+ :param jaas_override_variables: A dict of variables to be used in the jaas.conf template file
+ """
+ super(VerifiableShareConsumer, self).__init__(context, num_nodes)
+ self.log_level = log_level
+ self.kafka = kafka
+ self.topic = topic
+ self.group_id = group_id
+ self.offset_reset_strategy = offset_reset_strategy
+ self.max_messages = max_messages
+ self.acknowledgement_mode = acknowledgement_mode
+ self.prop_file = ""
+ self.stop_timeout_sec = stop_timeout_sec
+ self.on_record_consumed = on_record_consumed
+
+ self.event_handlers = {}
+ self.jaas_override_variables = jaas_override_variables or {}
+
+ self.total_records_consumed = 0
+ self.total_records_acknowledged = 0
+ self.total_records_acknowledged_failed = 0
+ self.consumed_records_offsets = set()
+ # self.consumed_more_than_once = []
+
+ self.acknowledged_records_offsets = set()
+ # self.acknowledged_more_than_once = []
+ self.is_offset_reset_strategy_set = False
+
+ for node in self.nodes:
+ node.version = version
+
+ def java_class_name(self):
+ return "VerifiableShareConsumer"
+
+ def create_event_handler(self, idx, node):
+ return ShareConsumerEventHandler(node, idx)
+
+ def _worker(self, idx, node):
+ with self.lock:
+ self.event_handlers[node] = self.create_event_handler(idx, node)
+ handler = self.event_handlers[node]
+
+ node.account.ssh("mkdir -p %s" % VerifiableShareConsumer.PERSISTENT_ROOT, allow_fail=False)
+
+ # Create and upload log properties
+ log_config = self.render('tools_log4j.properties', log_file=VerifiableShareConsumer.LOG_FILE)
Review Comment:
Hi, I have made the required changes in the latest commit, thanks !
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Infrastructure for system tests for the new share consumer client [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1919972479
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,630 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "offsetResetStrategy" })
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "count", "partitions" })
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "count", "partitions", "success", "error" })
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count = count;
+ this.partitions = partitions;
+ this.error = error;
+ this.success = success;
+ }
+
+ @Override
+ public String name() {
+ return "offsets_acknowledged";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<AcknowledgedData> partitions() {
+ return partitions;
+ }
+
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String error() {
+ return error;
+ }
+
+ @JsonProperty
+ public boolean success() {
+ return success;
+ }
+
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" })
+ public static class RecordData extends ShareConsumerEvent {
+
+ private final ConsumerRecord<String, String> record;
+
+ public RecordData(ConsumerRecord<String, String> record) {
+ this.record = record;
+ }
+
+ @Override
+ public String name() {
+ return "record_data";
+ }
+
+ @JsonProperty
+ public String topic() {
+ return record.topic();
+ }
+
+ @JsonProperty
+ public int partition() {
+ return record.partition();
+ }
+
+ @JsonProperty
Review Comment:
Key and value are both marked as JSON properties, but they're not in the JSON property order. This doesn't seem intentional. Please resolve.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Infrastructure for system tests for the new share consumer client [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #18209:
URL: https://github.com/apache/kafka/pull/18209#issuecomment-2602690747
This resolves KAFKA-17968.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Infrastructure for system tests for the new share consumer client [kafka]
Posted by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org>.
chirag-wadhwa5 commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1918716264
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH
+
+class ConsumerState:
+ Started = 1
+ Dead = 2
+
+class ShareConsumerEventHandler(object):
+
+ def __init__(self, node, idx, state=ConsumerState.Dead):
+ self.node = node
+ self.idx = idx
+ self.total_consumed = 0
+ self.total_acknowledged = 0
+ self.total_acknowledged_failed = 0
+ self.consumed_per_partition = {}
+ self.acknowledged_per_partition = {}
+ self.acknowledged_per_partition_failed = {}
+ self.state = state
+
+ def handle_shutdown_complete(self, node=None, logger=None):
+ self.state = ConsumerState.Dead
+ if node is not None and logger is not None:
+ logger.debug("Shut down %s" % node.account.hostname)
+
+ def handle_startup_complete(self, node, logger):
+ self.state = ConsumerState.Started
+ logger.debug("Started %s" % node.account.hostname)
+
+ def handle_offsets_acknowledged(self, event, node, logger):
+ if event["success"]:
+ self.total_acknowledged += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition[topic_partition] = self.acknowledged_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
Review Comment:
Thanks for the review. Currently, if we see, the "offsets_acknowledged" event is emitted by VerifiableShareConsumer when the acknowledgement commit callback is being called for the acknowledgements. The share consumer client is written in such a way that internally, this commit callback will always be called for a single topic partition data only, and the accompanying exception is provided with it. So if there is a case where SHARE_ACKNOWLEDGE response yields a result for multiple topic partitions where some partition are successful and others are not, then the callback will be called individually for those partitions, along with the exceptions in case of failed partitions. Thus, the event "offsets_acknowledged" is also emitted always for a single topic partition only. If there is an exception, the "success" value is set as false, otherwise it is set as true. I hope this clarifies everything !
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] First draft of system tests for the new share consumer client [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1887272452
##########
checkstyle/suppressions.xml:
##########
@@ -288,6 +288,8 @@
files="VerifiableConsumer.java"/>
<suppress id="dontUseSystemExit"
files="VerifiableProducer.java"/>
+ <suppress id="dontUseSystemExit"
+ files="VerifiableShareGroup.java"/>
Review Comment:
`VerifiableShareGroup` doesn't exist. Do you mean `VerifiableShareConsumer`?
##########
tests/kafkatest/services/verifiable_client.py:
##########
@@ -142,10 +142,10 @@
def create_verifiable_client_implementation(context, parent):
"""Factory for generating a verifiable client implementation class instance
- :param parent: parent class instance, either VerifiableConsumer or VerifiableProducer
+ :param parent: parent class instance, either VerifiableConsumer, VerifiableProducer or VerifiableShareGroup
Review Comment:
These references to `VerifiableShareGroup` and `ShareGroup` should probably be `VerifiableShareConsumer` and `ShareConsumer`.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] First draft of system tests for the new share consumer client [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1900783236
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,634 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private Integer totalConsumed = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count=count;
+ this.partitions = partitions;
+ this.error = error;
+ this.success = success;
+ }
+
+ @Override
+ public String name() {
+ return "offsets_acknowledged";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<AcknowledgedData> partitions() {
+ return partitions;
+ }
+
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String error() {
+ return error;
+ }
+
+ @JsonProperty
+ public boolean success() {
+ return success;
+ }
+
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" })
+ public static class RecordData extends ShareConsumerEvent {
+
+ private final ConsumerRecord<String, String> record;
+
+ public RecordData(ConsumerRecord<String, String> record) {
+ this.record = record;
+ }
+
+ @Override
+ public String name() {
+ return "record_data";
+ }
+
+ @JsonProperty
+ public String topic() {
+ return record.topic();
+ }
+
+ @JsonProperty
+ public int partition() {
+ return record.partition();
+ }
+
+ @JsonProperty
+ public String key() {
+ return record.key();
+ }
+
+ @JsonProperty
+ public String value() {
+ return record.value();
+ }
+
+ @JsonProperty
+ public long offset() {
+ return record.offset();
+ }
+
+ }
+
+ public VerifiableShareConsumer(KafkaShareConsumer<String, String> consumer,
+ PrintStream out,
+ Integer maxMessages,
+ String topic,
+ AcknowledgementMode acknowledgementMode,
+ String offsetResetStrategy,
+ String brokerHostandPort,
+ String groupId,
+ Boolean verbose) {
+ this.out = out;
+ this.consumer = consumer;
+ this.topic = topic;
+ this.acknowledgementMode = acknowledgementMode;
+ this.offsetResetStrategy = offsetResetStrategy;
+ this.verbose = verbose;
+ this.maxMessages = maxMessages;
+ this.brokerHostandPort = brokerHostandPort;
+ this.groupId = groupId;
+ addKafkaSerializerModule();
+ }
+
+ private void addKafkaSerializerModule() {
+ SimpleModule kafka = new SimpleModule();
+ kafka.addSerializer(TopicPartition.class, new JsonSerializer<>() {
+ @Override
+ public void serialize(TopicPartition tp, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+ gen.writeStartObject();
+ gen.writeObjectField("topic", tp.topic());
+ gen.writeObjectField("partition", tp.partition());
+ gen.writeEndObject();
+ }
+ });
+ mapper.registerModule(kafka);
+ }
+
+ private void onRecordsReceived(ConsumerRecords<String, String> records) {
+
+ List<RecordSetSummary> summaries = new ArrayList<>();
+ for (TopicPartition tp : records.partitions()) {
+ List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
+
+ if (partitionRecords.isEmpty())
+ continue;
+
+ Set<Long> partitionOffsets = new HashSet<>();
+
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ partitionOffsets.add(record.offset());
+ }
+
+ summaries.add(new RecordSetSummary(tp.topic(), tp.partition(), partitionOffsets));
+
+ if (this.verbose) {
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ printJson(new RecordData(record));
+ }
+ }
+
+ }
+
+ totalConsumed += records.count();
+ printJson(new RecordsConsumed(records.count(), summaries));
+ }
+
+ @Override
+ public void onComplete(Map<TopicIdPartition, Set<Long>> offsetsMap, Exception exception) {
+ List<AcknowledgedData> acknowledgedOffsets = new ArrayList<>();
+ int totalAcknowledged = 0;
+ for (Map.Entry<TopicIdPartition, Set<Long>> offsetEntry : offsetsMap.entrySet()) {
+ TopicIdPartition tp = offsetEntry.getKey();
+ acknowledgedOffsets.add(new AcknowledgedData(tp.topic(), tp.partition(), offsetEntry.getValue()));
+ totalAcknowledged += offsetEntry.getValue().size();
+ }
+
+ boolean success = true;
+ String error = null;
+ if (exception != null) {
+ success = false;
+ error = exception.getMessage();
+ }
+ printJson(new OffsetsAcknowledged(totalAcknowledged, acknowledgedOffsets, error, success));
+ if (success) {
+ this.totalAcknowledged += totalAcknowledged;
+ }
+ }
+
+
+ public void run() {
+ try {
+ printJson(new StartupComplete());
+
+ if (!Objects.equals(offsetResetStrategy, "")) {
+ ShareGroupAutoOffsetResetStrategy offsetResetStrategy =
+ ShareGroupAutoOffsetResetStrategy.fromString(this.offsetResetStrategy);
+
+ Properties adminClientProps = new Properties();
+ adminClientProps.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerHostandPort);
+
+ Admin adminClient = Admin.create(adminClientProps);
+
+ ConfigResource configResource = new ConfigResource(ConfigResource.Type.GROUP, groupId);
+ Map<ConfigResource, Collection<AlterConfigOp>> alterEntries = new HashMap<>();
+ alterEntries.put(configResource, List.of(new AlterConfigOp(new ConfigEntry(
+ GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, offsetResetStrategy.type().toString()), AlterConfigOp.OpType.SET)));
+ AlterConfigsOptions alterOptions = new AlterConfigsOptions();
+
+ // Setting the share group auto offset reset strategy
+ adminClient.incrementalAlterConfigs(alterEntries, alterOptions)
+ .all()
+ .get(60, TimeUnit.SECONDS);
+
+ printJson(new OffsetResetStrategySet(offsetResetStrategy.type().toString()));
+ }
+
+ consumer.subscribe(Collections.singleton(this.topic));
+ consumer.setAcknowledgementCommitCallback(this);
+ while (!(maxMessages >= 0 && totalAcknowledged >= maxMessages)) {
+ ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5000));
+ if (!records.isEmpty()) {
+ onRecordsReceived(records);
+
+ if (acknowledgementMode == AcknowledgementMode.ASYNC) {
+ consumer.commitAsync();
+ } else if (acknowledgementMode == AcknowledgementMode.SYNC) {
+ Map<TopicIdPartition, Optional<KafkaException>> result = consumer.commitSync();
+ for (Map.Entry<TopicIdPartition, Optional<KafkaException>> resultEntry : result.entrySet()) {
+ if (resultEntry.getValue().isPresent()) {
+ log.error("Failed to commit offset synchronously for topic partition: {}", resultEntry.getKey());
+ }
+ }
+ }
+ }
+ }
+ } catch (WakeupException e) {
+ out.println("caught wakeup exception: " + e);
+ // ignore, we are closing
+ log.trace("Caught WakeupException because share consumer is shutdown, ignore and terminate.", e);
+ } catch (Throwable t) {
+ out.println("caught throwable exception: " + t);
+ // Log the error, so it goes to the service log and not stdout
+ log.error("Error during processing, terminating share consumer process: ", t);
+ } finally {
+ out.println("Total records Consumed: " + totalConsumed.toString());
+ out.println("Total records Acknowledged: " + totalAcknowledged.toString());
+ consumer.close();
+ printJson(new ShutdownComplete());
+ shutdownLatch.countDown();
+ }
+ }
+
+ public void close() {
+ boolean interrupted = false;
+ try {
+ consumer.wakeup();
+ while (true) {
+ try {
+ shutdownLatch.await();
+ return;
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ } finally {
+ if (interrupted)
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ protected synchronized void printJson(Object data) {
+ try {
+ out.println(mapper.writeValueAsString(data));
+ } catch (JsonProcessingException e) {
+ out.println("Bad data can't be written as json: " + e.getMessage());
+ }
+ }
+
+ public enum AcknowledgementMode {
+ AUTO, ASYNC, SYNC;
+
+ @Override
+ public String toString() {
+ return super.toString().toLowerCase(Locale.ROOT);
+ }
+ }
+
+ private static ArgumentParser argParser() {
+ ArgumentParser parser = ArgumentParsers
+ .newArgumentParser("verifiable-share-group")
Review Comment:
"verifiable-share-consumer"
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,634 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private Integer totalConsumed = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count=count;
+ this.partitions = partitions;
+ this.error = error;
+ this.success = success;
+ }
+
+ @Override
+ public String name() {
+ return "offsets_acknowledged";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<AcknowledgedData> partitions() {
+ return partitions;
+ }
+
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String error() {
+ return error;
+ }
+
+ @JsonProperty
+ public boolean success() {
+ return success;
+ }
+
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" })
+ public static class RecordData extends ShareConsumerEvent {
+
+ private final ConsumerRecord<String, String> record;
+
+ public RecordData(ConsumerRecord<String, String> record) {
+ this.record = record;
+ }
+
+ @Override
+ public String name() {
+ return "record_data";
+ }
+
+ @JsonProperty
+ public String topic() {
+ return record.topic();
+ }
+
+ @JsonProperty
+ public int partition() {
+ return record.partition();
+ }
+
+ @JsonProperty
+ public String key() {
+ return record.key();
+ }
+
+ @JsonProperty
+ public String value() {
+ return record.value();
+ }
+
+ @JsonProperty
+ public long offset() {
+ return record.offset();
+ }
+
+ }
+
+ public VerifiableShareConsumer(KafkaShareConsumer<String, String> consumer,
+ PrintStream out,
+ Integer maxMessages,
+ String topic,
+ AcknowledgementMode acknowledgementMode,
+ String offsetResetStrategy,
+ String brokerHostandPort,
+ String groupId,
+ Boolean verbose) {
+ this.out = out;
+ this.consumer = consumer;
+ this.topic = topic;
+ this.acknowledgementMode = acknowledgementMode;
+ this.offsetResetStrategy = offsetResetStrategy;
+ this.verbose = verbose;
+ this.maxMessages = maxMessages;
+ this.brokerHostandPort = brokerHostandPort;
+ this.groupId = groupId;
+ addKafkaSerializerModule();
+ }
+
+ private void addKafkaSerializerModule() {
+ SimpleModule kafka = new SimpleModule();
+ kafka.addSerializer(TopicPartition.class, new JsonSerializer<>() {
+ @Override
+ public void serialize(TopicPartition tp, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+ gen.writeStartObject();
+ gen.writeObjectField("topic", tp.topic());
+ gen.writeObjectField("partition", tp.partition());
+ gen.writeEndObject();
+ }
+ });
+ mapper.registerModule(kafka);
+ }
+
+ private void onRecordsReceived(ConsumerRecords<String, String> records) {
+
+ List<RecordSetSummary> summaries = new ArrayList<>();
+ for (TopicPartition tp : records.partitions()) {
+ List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
+
+ if (partitionRecords.isEmpty())
+ continue;
+
+ Set<Long> partitionOffsets = new HashSet<>();
+
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ partitionOffsets.add(record.offset());
+ }
+
+ summaries.add(new RecordSetSummary(tp.topic(), tp.partition(), partitionOffsets));
+
+ if (this.verbose) {
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ printJson(new RecordData(record));
+ }
+ }
+
+ }
+
+ totalConsumed += records.count();
+ printJson(new RecordsConsumed(records.count(), summaries));
+ }
+
+ @Override
+ public void onComplete(Map<TopicIdPartition, Set<Long>> offsetsMap, Exception exception) {
+ List<AcknowledgedData> acknowledgedOffsets = new ArrayList<>();
+ int totalAcknowledged = 0;
+ for (Map.Entry<TopicIdPartition, Set<Long>> offsetEntry : offsetsMap.entrySet()) {
+ TopicIdPartition tp = offsetEntry.getKey();
+ acknowledgedOffsets.add(new AcknowledgedData(tp.topic(), tp.partition(), offsetEntry.getValue()));
+ totalAcknowledged += offsetEntry.getValue().size();
+ }
+
+ boolean success = true;
+ String error = null;
+ if (exception != null) {
+ success = false;
+ error = exception.getMessage();
+ }
+ printJson(new OffsetsAcknowledged(totalAcknowledged, acknowledgedOffsets, error, success));
+ if (success) {
+ this.totalAcknowledged += totalAcknowledged;
+ }
+ }
+
+
+ public void run() {
+ try {
+ printJson(new StartupComplete());
+
+ if (!Objects.equals(offsetResetStrategy, "")) {
+ ShareGroupAutoOffsetResetStrategy offsetResetStrategy =
+ ShareGroupAutoOffsetResetStrategy.fromString(this.offsetResetStrategy);
+
+ Properties adminClientProps = new Properties();
+ adminClientProps.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerHostandPort);
+
+ Admin adminClient = Admin.create(adminClientProps);
+
+ ConfigResource configResource = new ConfigResource(ConfigResource.Type.GROUP, groupId);
+ Map<ConfigResource, Collection<AlterConfigOp>> alterEntries = new HashMap<>();
+ alterEntries.put(configResource, List.of(new AlterConfigOp(new ConfigEntry(
+ GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, offsetResetStrategy.type().toString()), AlterConfigOp.OpType.SET)));
+ AlterConfigsOptions alterOptions = new AlterConfigsOptions();
+
+ // Setting the share group auto offset reset strategy
+ adminClient.incrementalAlterConfigs(alterEntries, alterOptions)
+ .all()
+ .get(60, TimeUnit.SECONDS);
+
+ printJson(new OffsetResetStrategySet(offsetResetStrategy.type().toString()));
+ }
+
+ consumer.subscribe(Collections.singleton(this.topic));
+ consumer.setAcknowledgementCommitCallback(this);
+ while (!(maxMessages >= 0 && totalAcknowledged >= maxMessages)) {
+ ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5000));
+ if (!records.isEmpty()) {
+ onRecordsReceived(records);
+
+ if (acknowledgementMode == AcknowledgementMode.ASYNC) {
+ consumer.commitAsync();
+ } else if (acknowledgementMode == AcknowledgementMode.SYNC) {
+ Map<TopicIdPartition, Optional<KafkaException>> result = consumer.commitSync();
+ for (Map.Entry<TopicIdPartition, Optional<KafkaException>> resultEntry : result.entrySet()) {
+ if (resultEntry.getValue().isPresent()) {
+ log.error("Failed to commit offset synchronously for topic partition: {}", resultEntry.getKey());
+ }
+ }
+ }
+ }
+ }
+ } catch (WakeupException e) {
+ out.println("caught wakeup exception: " + e);
+ // ignore, we are closing
+ log.trace("Caught WakeupException because share consumer is shutdown, ignore and terminate.", e);
+ } catch (Throwable t) {
+ out.println("caught throwable exception: " + t);
+ // Log the error, so it goes to the service log and not stdout
+ log.error("Error during processing, terminating share consumer process: ", t);
+ } finally {
+ out.println("Total records Consumed: " + totalConsumed.toString());
+ out.println("Total records Acknowledged: " + totalAcknowledged.toString());
+ consumer.close();
+ printJson(new ShutdownComplete());
+ shutdownLatch.countDown();
+ }
+ }
+
+ public void close() {
+ boolean interrupted = false;
+ try {
+ consumer.wakeup();
+ while (true) {
+ try {
+ shutdownLatch.await();
+ return;
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ } finally {
+ if (interrupted)
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ protected synchronized void printJson(Object data) {
+ try {
+ out.println(mapper.writeValueAsString(data));
+ } catch (JsonProcessingException e) {
+ out.println("Bad data can't be written as json: " + e.getMessage());
+ }
+ }
+
+ public enum AcknowledgementMode {
+ AUTO, ASYNC, SYNC;
+
+ @Override
+ public String toString() {
+ return super.toString().toLowerCase(Locale.ROOT);
+ }
+ }
+
+ private static ArgumentParser argParser() {
+ ArgumentParser parser = ArgumentParsers
+ .newArgumentParser("verifiable-share-group")
+ .defaultHelp(true)
+ .description("This tool creates a share group and consumes messages from a specific topic and emits share consumer events (e.g. share consumer startup, received messages, and offsets acknowledged) as JSON objects to STDOUT.");
+ MutuallyExclusiveGroup connectionGroup = parser.addMutuallyExclusiveGroup("Connection Group")
+ .description("Group of arguments for connection to brokers")
+ .required(true);
+ connectionGroup.addArgument("--bootstrap-server")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .metavar("HOST1:PORT1[,HOST2:PORT2[...]]")
+ .dest("bootstrapServer")
+ .help("The server(s) to connect to. Comma-separated list of Kafka brokers in the form HOST1:PORT1,HOST2:PORT2,...");
+
+ parser.addArgument("--topic")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .metavar("TOPIC")
+ .help("Consumes messages from this topic.");
+
+ parser.addArgument("--group-id")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .metavar("GROUP_ID")
+ .dest("groupId")
+ .help("The groupId shared among members of the consumer group");
+
+ parser.addArgument("--max-messages")
+ .action(store())
+ .required(false)
+ .type(Integer.class)
+ .setDefault(-1)
+ .metavar("MAX-MESSAGES")
+ .dest("maxMessages")
+ .help("Consume this many messages. If -1 (the default), the share consumers will consume until the process is killed externally");
+
+ parser.addArgument("--verbose")
+ .action(storeTrue())
+ .type(Boolean.class)
+ .metavar("VERBOSE")
+ .help("Enable to log individual consumed records");
+
+ parser.addArgument("--acknowledgement-mode")
+ .action(store())
+ .required(false)
+ .setDefault("auto")
+ .type(String.class)
+ .dest("acknowledgementMode")
+ .help("Acknowledgement mode for the share consumers (must be either 'auto', 'sync' or 'async')");
+
+ parser.addArgument("--offset-reset-strategy")
+ .action(store())
+ .required(false)
+ .setDefault("")
+ .type(String.class)
+ .dest("offsetResetStrategy")
+ .help("Set share group reset strategy (must be either 'earliest' or 'latest'");
Review Comment:
nit: Missing ')' int he string.
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,634 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private Integer totalConsumed = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count=count;
+ this.partitions = partitions;
+ this.error = error;
+ this.success = success;
+ }
+
+ @Override
+ public String name() {
+ return "offsets_acknowledged";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<AcknowledgedData> partitions() {
+ return partitions;
+ }
+
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String error() {
+ return error;
+ }
+
+ @JsonProperty
+ public boolean success() {
+ return success;
+ }
+
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" })
+ public static class RecordData extends ShareConsumerEvent {
+
+ private final ConsumerRecord<String, String> record;
+
+ public RecordData(ConsumerRecord<String, String> record) {
+ this.record = record;
+ }
+
+ @Override
+ public String name() {
+ return "record_data";
+ }
+
+ @JsonProperty
+ public String topic() {
+ return record.topic();
+ }
+
+ @JsonProperty
+ public int partition() {
+ return record.partition();
+ }
+
+ @JsonProperty
+ public String key() {
+ return record.key();
+ }
+
+ @JsonProperty
+ public String value() {
+ return record.value();
+ }
+
+ @JsonProperty
+ public long offset() {
+ return record.offset();
+ }
+
+ }
+
+ public VerifiableShareConsumer(KafkaShareConsumer<String, String> consumer,
+ PrintStream out,
+ Integer maxMessages,
+ String topic,
+ AcknowledgementMode acknowledgementMode,
+ String offsetResetStrategy,
+ String brokerHostandPort,
+ String groupId,
+ Boolean verbose) {
+ this.out = out;
+ this.consumer = consumer;
+ this.topic = topic;
+ this.acknowledgementMode = acknowledgementMode;
+ this.offsetResetStrategy = offsetResetStrategy;
+ this.verbose = verbose;
+ this.maxMessages = maxMessages;
+ this.brokerHostandPort = brokerHostandPort;
+ this.groupId = groupId;
+ addKafkaSerializerModule();
+ }
+
+ private void addKafkaSerializerModule() {
+ SimpleModule kafka = new SimpleModule();
+ kafka.addSerializer(TopicPartition.class, new JsonSerializer<>() {
+ @Override
+ public void serialize(TopicPartition tp, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+ gen.writeStartObject();
+ gen.writeObjectField("topic", tp.topic());
+ gen.writeObjectField("partition", tp.partition());
+ gen.writeEndObject();
+ }
+ });
+ mapper.registerModule(kafka);
+ }
+
+ private void onRecordsReceived(ConsumerRecords<String, String> records) {
+
+ List<RecordSetSummary> summaries = new ArrayList<>();
+ for (TopicPartition tp : records.partitions()) {
+ List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
+
+ if (partitionRecords.isEmpty())
+ continue;
+
+ Set<Long> partitionOffsets = new HashSet<>();
+
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ partitionOffsets.add(record.offset());
+ }
+
+ summaries.add(new RecordSetSummary(tp.topic(), tp.partition(), partitionOffsets));
+
+ if (this.verbose) {
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ printJson(new RecordData(record));
+ }
+ }
+
+ }
+
+ totalConsumed += records.count();
+ printJson(new RecordsConsumed(records.count(), summaries));
+ }
+
+ @Override
+ public void onComplete(Map<TopicIdPartition, Set<Long>> offsetsMap, Exception exception) {
+ List<AcknowledgedData> acknowledgedOffsets = new ArrayList<>();
+ int totalAcknowledged = 0;
+ for (Map.Entry<TopicIdPartition, Set<Long>> offsetEntry : offsetsMap.entrySet()) {
+ TopicIdPartition tp = offsetEntry.getKey();
+ acknowledgedOffsets.add(new AcknowledgedData(tp.topic(), tp.partition(), offsetEntry.getValue()));
+ totalAcknowledged += offsetEntry.getValue().size();
+ }
+
+ boolean success = true;
+ String error = null;
+ if (exception != null) {
+ success = false;
+ error = exception.getMessage();
+ }
+ printJson(new OffsetsAcknowledged(totalAcknowledged, acknowledgedOffsets, error, success));
+ if (success) {
+ this.totalAcknowledged += totalAcknowledged;
+ }
+ }
+
+
+ public void run() {
+ try {
+ printJson(new StartupComplete());
+
+ if (!Objects.equals(offsetResetStrategy, "")) {
+ ShareGroupAutoOffsetResetStrategy offsetResetStrategy =
+ ShareGroupAutoOffsetResetStrategy.fromString(this.offsetResetStrategy);
+
+ Properties adminClientProps = new Properties();
+ adminClientProps.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerHostandPort);
+
+ Admin adminClient = Admin.create(adminClientProps);
+
+ ConfigResource configResource = new ConfigResource(ConfigResource.Type.GROUP, groupId);
+ Map<ConfigResource, Collection<AlterConfigOp>> alterEntries = new HashMap<>();
+ alterEntries.put(configResource, List.of(new AlterConfigOp(new ConfigEntry(
+ GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, offsetResetStrategy.type().toString()), AlterConfigOp.OpType.SET)));
+ AlterConfigsOptions alterOptions = new AlterConfigsOptions();
+
+ // Setting the share group auto offset reset strategy
+ adminClient.incrementalAlterConfigs(alterEntries, alterOptions)
+ .all()
+ .get(60, TimeUnit.SECONDS);
+
+ printJson(new OffsetResetStrategySet(offsetResetStrategy.type().toString()));
+ }
+
+ consumer.subscribe(Collections.singleton(this.topic));
+ consumer.setAcknowledgementCommitCallback(this);
+ while (!(maxMessages >= 0 && totalAcknowledged >= maxMessages)) {
+ ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5000));
+ if (!records.isEmpty()) {
+ onRecordsReceived(records);
+
+ if (acknowledgementMode == AcknowledgementMode.ASYNC) {
+ consumer.commitAsync();
+ } else if (acknowledgementMode == AcknowledgementMode.SYNC) {
+ Map<TopicIdPartition, Optional<KafkaException>> result = consumer.commitSync();
+ for (Map.Entry<TopicIdPartition, Optional<KafkaException>> resultEntry : result.entrySet()) {
+ if (resultEntry.getValue().isPresent()) {
+ log.error("Failed to commit offset synchronously for topic partition: {}", resultEntry.getKey());
+ }
+ }
+ }
+ }
+ }
+ } catch (WakeupException e) {
+ out.println("caught wakeup exception: " + e);
+ // ignore, we are closing
+ log.trace("Caught WakeupException because share consumer is shutdown, ignore and terminate.", e);
+ } catch (Throwable t) {
+ out.println("caught throwable exception: " + t);
+ // Log the error, so it goes to the service log and not stdout
+ log.error("Error during processing, terminating share consumer process: ", t);
+ } finally {
+ out.println("Total records Consumed: " + totalConsumed.toString());
Review Comment:
If we want to write out these pieces of information, I suggest a better way is appropriate. I'd remove these two printlns.
##########
tests/kafkatest/services/kafka/templates/kafka.properties:
##########
@@ -126,3 +126,11 @@ offsets.topic.num.partitions={{ num_nodes }}
offsets.topic.replication.factor={{ 3 if num_nodes > 3 else num_nodes }}
# Set to a low, but non-zero value to exercise this path without making tests much slower
group.initial.rebalance.delay.ms=100
+
+group.coordinator.rebalance.protocols=classic,consumer,share
Review Comment:
There needs to be a better way to manage the configs here. Changing this file changes the configs for all of the tests, and that doesn't seem right when KIP-932 is still in early access. I think you should not include this change in the PR.
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,634 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private Integer totalConsumed = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count=count;
Review Comment:
nit: spacing
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,634 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private Integer totalConsumed = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count=count;
+ this.partitions = partitions;
+ this.error = error;
+ this.success = success;
+ }
+
+ @Override
+ public String name() {
+ return "offsets_acknowledged";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<AcknowledgedData> partitions() {
+ return partitions;
+ }
+
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String error() {
+ return error;
+ }
+
+ @JsonProperty
+ public boolean success() {
+ return success;
+ }
+
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" })
+ public static class RecordData extends ShareConsumerEvent {
+
+ private final ConsumerRecord<String, String> record;
+
+ public RecordData(ConsumerRecord<String, String> record) {
+ this.record = record;
+ }
+
+ @Override
+ public String name() {
+ return "record_data";
+ }
+
+ @JsonProperty
+ public String topic() {
+ return record.topic();
+ }
+
+ @JsonProperty
+ public int partition() {
+ return record.partition();
+ }
+
+ @JsonProperty
+ public String key() {
+ return record.key();
+ }
+
+ @JsonProperty
+ public String value() {
+ return record.value();
+ }
+
+ @JsonProperty
+ public long offset() {
+ return record.offset();
+ }
+
+ }
+
+ public VerifiableShareConsumer(KafkaShareConsumer<String, String> consumer,
+ PrintStream out,
+ Integer maxMessages,
+ String topic,
+ AcknowledgementMode acknowledgementMode,
+ String offsetResetStrategy,
+ String brokerHostandPort,
+ String groupId,
+ Boolean verbose) {
+ this.out = out;
+ this.consumer = consumer;
+ this.topic = topic;
+ this.acknowledgementMode = acknowledgementMode;
+ this.offsetResetStrategy = offsetResetStrategy;
+ this.verbose = verbose;
+ this.maxMessages = maxMessages;
+ this.brokerHostandPort = brokerHostandPort;
+ this.groupId = groupId;
+ addKafkaSerializerModule();
+ }
+
+ private void addKafkaSerializerModule() {
+ SimpleModule kafka = new SimpleModule();
+ kafka.addSerializer(TopicPartition.class, new JsonSerializer<>() {
+ @Override
+ public void serialize(TopicPartition tp, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+ gen.writeStartObject();
+ gen.writeObjectField("topic", tp.topic());
+ gen.writeObjectField("partition", tp.partition());
+ gen.writeEndObject();
+ }
+ });
+ mapper.registerModule(kafka);
+ }
+
+ private void onRecordsReceived(ConsumerRecords<String, String> records) {
+
+ List<RecordSetSummary> summaries = new ArrayList<>();
+ for (TopicPartition tp : records.partitions()) {
+ List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
+
+ if (partitionRecords.isEmpty())
+ continue;
+
+ Set<Long> partitionOffsets = new HashSet<>();
+
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ partitionOffsets.add(record.offset());
+ }
+
+ summaries.add(new RecordSetSummary(tp.topic(), tp.partition(), partitionOffsets));
+
+ if (this.verbose) {
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ printJson(new RecordData(record));
+ }
+ }
+
+ }
+
+ totalConsumed += records.count();
+ printJson(new RecordsConsumed(records.count(), summaries));
+ }
+
+ @Override
+ public void onComplete(Map<TopicIdPartition, Set<Long>> offsetsMap, Exception exception) {
+ List<AcknowledgedData> acknowledgedOffsets = new ArrayList<>();
+ int totalAcknowledged = 0;
+ for (Map.Entry<TopicIdPartition, Set<Long>> offsetEntry : offsetsMap.entrySet()) {
+ TopicIdPartition tp = offsetEntry.getKey();
+ acknowledgedOffsets.add(new AcknowledgedData(tp.topic(), tp.partition(), offsetEntry.getValue()));
+ totalAcknowledged += offsetEntry.getValue().size();
+ }
+
+ boolean success = true;
+ String error = null;
+ if (exception != null) {
+ success = false;
+ error = exception.getMessage();
+ }
+ printJson(new OffsetsAcknowledged(totalAcknowledged, acknowledgedOffsets, error, success));
+ if (success) {
+ this.totalAcknowledged += totalAcknowledged;
+ }
+ }
+
+
+ public void run() {
+ try {
+ printJson(new StartupComplete());
+
+ if (!Objects.equals(offsetResetStrategy, "")) {
+ ShareGroupAutoOffsetResetStrategy offsetResetStrategy =
+ ShareGroupAutoOffsetResetStrategy.fromString(this.offsetResetStrategy);
+
+ Properties adminClientProps = new Properties();
+ adminClientProps.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerHostandPort);
+
+ Admin adminClient = Admin.create(adminClientProps);
+
+ ConfigResource configResource = new ConfigResource(ConfigResource.Type.GROUP, groupId);
+ Map<ConfigResource, Collection<AlterConfigOp>> alterEntries = new HashMap<>();
+ alterEntries.put(configResource, List.of(new AlterConfigOp(new ConfigEntry(
+ GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, offsetResetStrategy.type().toString()), AlterConfigOp.OpType.SET)));
+ AlterConfigsOptions alterOptions = new AlterConfigsOptions();
+
+ // Setting the share group auto offset reset strategy
+ adminClient.incrementalAlterConfigs(alterEntries, alterOptions)
+ .all()
+ .get(60, TimeUnit.SECONDS);
+
+ printJson(new OffsetResetStrategySet(offsetResetStrategy.type().toString()));
+ }
+
+ consumer.subscribe(Collections.singleton(this.topic));
+ consumer.setAcknowledgementCommitCallback(this);
+ while (!(maxMessages >= 0 && totalAcknowledged >= maxMessages)) {
+ ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5000));
+ if (!records.isEmpty()) {
+ onRecordsReceived(records);
+
+ if (acknowledgementMode == AcknowledgementMode.ASYNC) {
+ consumer.commitAsync();
+ } else if (acknowledgementMode == AcknowledgementMode.SYNC) {
+ Map<TopicIdPartition, Optional<KafkaException>> result = consumer.commitSync();
+ for (Map.Entry<TopicIdPartition, Optional<KafkaException>> resultEntry : result.entrySet()) {
+ if (resultEntry.getValue().isPresent()) {
+ log.error("Failed to commit offset synchronously for topic partition: {}", resultEntry.getKey());
+ }
+ }
+ }
+ }
+ }
+ } catch (WakeupException e) {
+ out.println("caught wakeup exception: " + e);
+ // ignore, we are closing
+ log.trace("Caught WakeupException because share consumer is shutdown, ignore and terminate.", e);
+ } catch (Throwable t) {
+ out.println("caught throwable exception: " + t);
Review Comment:
And this one.
##########
run_tests.sh:
##########
@@ -0,0 +1,35 @@
+#!/bin/bash
Review Comment:
This file seems like a script you could use locally to run using docker. I don't think it should be part of a PR.
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,340 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH, V_2_3_0, V_2_3_1, V_3_7_0, V_4_0_0
Review Comment:
I think this should be applicable for DEV_BRANCH and V_4_1_0 only.
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,634 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private Integer totalConsumed = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count=count;
+ this.partitions = partitions;
+ this.error = error;
+ this.success = success;
+ }
+
+ @Override
+ public String name() {
+ return "offsets_acknowledged";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<AcknowledgedData> partitions() {
+ return partitions;
+ }
+
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String error() {
+ return error;
+ }
+
+ @JsonProperty
+ public boolean success() {
+ return success;
+ }
+
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" })
+ public static class RecordData extends ShareConsumerEvent {
+
+ private final ConsumerRecord<String, String> record;
+
+ public RecordData(ConsumerRecord<String, String> record) {
+ this.record = record;
+ }
+
+ @Override
+ public String name() {
+ return "record_data";
+ }
+
+ @JsonProperty
+ public String topic() {
+ return record.topic();
+ }
+
+ @JsonProperty
+ public int partition() {
+ return record.partition();
+ }
+
+ @JsonProperty
+ public String key() {
+ return record.key();
+ }
+
+ @JsonProperty
+ public String value() {
+ return record.value();
+ }
+
+ @JsonProperty
+ public long offset() {
+ return record.offset();
+ }
+
+ }
+
+ public VerifiableShareConsumer(KafkaShareConsumer<String, String> consumer,
+ PrintStream out,
+ Integer maxMessages,
+ String topic,
+ AcknowledgementMode acknowledgementMode,
+ String offsetResetStrategy,
+ String brokerHostandPort,
+ String groupId,
+ Boolean verbose) {
+ this.out = out;
+ this.consumer = consumer;
+ this.topic = topic;
+ this.acknowledgementMode = acknowledgementMode;
+ this.offsetResetStrategy = offsetResetStrategy;
+ this.verbose = verbose;
+ this.maxMessages = maxMessages;
+ this.brokerHostandPort = brokerHostandPort;
+ this.groupId = groupId;
+ addKafkaSerializerModule();
+ }
+
+ private void addKafkaSerializerModule() {
+ SimpleModule kafka = new SimpleModule();
+ kafka.addSerializer(TopicPartition.class, new JsonSerializer<>() {
+ @Override
+ public void serialize(TopicPartition tp, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+ gen.writeStartObject();
+ gen.writeObjectField("topic", tp.topic());
+ gen.writeObjectField("partition", tp.partition());
+ gen.writeEndObject();
+ }
+ });
+ mapper.registerModule(kafka);
+ }
+
+ private void onRecordsReceived(ConsumerRecords<String, String> records) {
+
+ List<RecordSetSummary> summaries = new ArrayList<>();
+ for (TopicPartition tp : records.partitions()) {
+ List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
+
+ if (partitionRecords.isEmpty())
+ continue;
+
+ Set<Long> partitionOffsets = new HashSet<>();
+
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ partitionOffsets.add(record.offset());
+ }
+
+ summaries.add(new RecordSetSummary(tp.topic(), tp.partition(), partitionOffsets));
+
+ if (this.verbose) {
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ printJson(new RecordData(record));
+ }
+ }
+
+ }
+
+ totalConsumed += records.count();
+ printJson(new RecordsConsumed(records.count(), summaries));
+ }
+
+ @Override
+ public void onComplete(Map<TopicIdPartition, Set<Long>> offsetsMap, Exception exception) {
+ List<AcknowledgedData> acknowledgedOffsets = new ArrayList<>();
+ int totalAcknowledged = 0;
+ for (Map.Entry<TopicIdPartition, Set<Long>> offsetEntry : offsetsMap.entrySet()) {
+ TopicIdPartition tp = offsetEntry.getKey();
+ acknowledgedOffsets.add(new AcknowledgedData(tp.topic(), tp.partition(), offsetEntry.getValue()));
+ totalAcknowledged += offsetEntry.getValue().size();
+ }
+
+ boolean success = true;
+ String error = null;
+ if (exception != null) {
+ success = false;
+ error = exception.getMessage();
+ }
+ printJson(new OffsetsAcknowledged(totalAcknowledged, acknowledgedOffsets, error, success));
+ if (success) {
+ this.totalAcknowledged += totalAcknowledged;
+ }
+ }
+
+
+ public void run() {
+ try {
+ printJson(new StartupComplete());
+
+ if (!Objects.equals(offsetResetStrategy, "")) {
+ ShareGroupAutoOffsetResetStrategy offsetResetStrategy =
+ ShareGroupAutoOffsetResetStrategy.fromString(this.offsetResetStrategy);
+
+ Properties adminClientProps = new Properties();
+ adminClientProps.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerHostandPort);
+
+ Admin adminClient = Admin.create(adminClientProps);
+
+ ConfigResource configResource = new ConfigResource(ConfigResource.Type.GROUP, groupId);
+ Map<ConfigResource, Collection<AlterConfigOp>> alterEntries = new HashMap<>();
+ alterEntries.put(configResource, List.of(new AlterConfigOp(new ConfigEntry(
+ GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, offsetResetStrategy.type().toString()), AlterConfigOp.OpType.SET)));
+ AlterConfigsOptions alterOptions = new AlterConfigsOptions();
+
+ // Setting the share group auto offset reset strategy
+ adminClient.incrementalAlterConfigs(alterEntries, alterOptions)
+ .all()
+ .get(60, TimeUnit.SECONDS);
+
+ printJson(new OffsetResetStrategySet(offsetResetStrategy.type().toString()));
+ }
+
+ consumer.subscribe(Collections.singleton(this.topic));
+ consumer.setAcknowledgementCommitCallback(this);
+ while (!(maxMessages >= 0 && totalAcknowledged >= maxMessages)) {
+ ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5000));
+ if (!records.isEmpty()) {
+ onRecordsReceived(records);
+
+ if (acknowledgementMode == AcknowledgementMode.ASYNC) {
+ consumer.commitAsync();
+ } else if (acknowledgementMode == AcknowledgementMode.SYNC) {
+ Map<TopicIdPartition, Optional<KafkaException>> result = consumer.commitSync();
+ for (Map.Entry<TopicIdPartition, Optional<KafkaException>> resultEntry : result.entrySet()) {
+ if (resultEntry.getValue().isPresent()) {
+ log.error("Failed to commit offset synchronously for topic partition: {}", resultEntry.getKey());
+ }
+ }
+ }
+ }
+ }
+ } catch (WakeupException e) {
+ out.println("caught wakeup exception: " + e);
Review Comment:
Remove this line please.
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,634 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private Integer totalConsumed = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count=count;
+ this.partitions = partitions;
+ this.error = error;
+ this.success = success;
+ }
+
+ @Override
+ public String name() {
+ return "offsets_acknowledged";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<AcknowledgedData> partitions() {
+ return partitions;
+ }
+
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String error() {
+ return error;
+ }
+
+ @JsonProperty
+ public boolean success() {
+ return success;
+ }
+
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" })
+ public static class RecordData extends ShareConsumerEvent {
+
+ private final ConsumerRecord<String, String> record;
+
+ public RecordData(ConsumerRecord<String, String> record) {
+ this.record = record;
+ }
+
+ @Override
+ public String name() {
+ return "record_data";
+ }
+
+ @JsonProperty
+ public String topic() {
+ return record.topic();
+ }
+
+ @JsonProperty
+ public int partition() {
+ return record.partition();
+ }
+
+ @JsonProperty
+ public String key() {
+ return record.key();
+ }
+
+ @JsonProperty
+ public String value() {
+ return record.value();
+ }
+
+ @JsonProperty
+ public long offset() {
+ return record.offset();
+ }
+
+ }
+
+ public VerifiableShareConsumer(KafkaShareConsumer<String, String> consumer,
+ PrintStream out,
+ Integer maxMessages,
+ String topic,
+ AcknowledgementMode acknowledgementMode,
+ String offsetResetStrategy,
+ String brokerHostandPort,
+ String groupId,
+ Boolean verbose) {
+ this.out = out;
+ this.consumer = consumer;
+ this.topic = topic;
+ this.acknowledgementMode = acknowledgementMode;
+ this.offsetResetStrategy = offsetResetStrategy;
+ this.verbose = verbose;
+ this.maxMessages = maxMessages;
+ this.brokerHostandPort = brokerHostandPort;
+ this.groupId = groupId;
+ addKafkaSerializerModule();
+ }
+
+ private void addKafkaSerializerModule() {
+ SimpleModule kafka = new SimpleModule();
+ kafka.addSerializer(TopicPartition.class, new JsonSerializer<>() {
+ @Override
+ public void serialize(TopicPartition tp, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+ gen.writeStartObject();
+ gen.writeObjectField("topic", tp.topic());
+ gen.writeObjectField("partition", tp.partition());
+ gen.writeEndObject();
+ }
+ });
+ mapper.registerModule(kafka);
+ }
+
+ private void onRecordsReceived(ConsumerRecords<String, String> records) {
+
+ List<RecordSetSummary> summaries = new ArrayList<>();
+ for (TopicPartition tp : records.partitions()) {
+ List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
+
+ if (partitionRecords.isEmpty())
+ continue;
+
+ Set<Long> partitionOffsets = new HashSet<>();
+
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ partitionOffsets.add(record.offset());
+ }
+
+ summaries.add(new RecordSetSummary(tp.topic(), tp.partition(), partitionOffsets));
+
+ if (this.verbose) {
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ printJson(new RecordData(record));
+ }
+ }
+
Review Comment:
nit: Extraneous line.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Infrastructure for system tests for the new share consumer client [kafka]
Posted by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org>.
chirag-wadhwa5 commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1918752625
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,627 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
Review Comment:
Thanks for the review. The classes `RecordSetSummary` and 'AcknowledgedData' extends PartitionData to inherit the topic and partition details for the respective consumption and acknowledgement event. If we see the ultimate Json string emitted by these classes, all of the class variables, even the inherited ones are printed, which are then read by the docker container where the system test is running for further processing. And the system test requires the topic partition information for further processing of the events. Do you have any other suggestion for this ?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] First draft of system tests for the new share consumer client [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1913136314
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
Review Comment:
nit: line break
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH
+
+class ConsumerState:
Review Comment:
nit: may be just `State` or `ShareConsumerState`
##########
tests/kafkatest/tests/client/share_consumer_test.py:
##########
@@ -0,0 +1,313 @@
+# 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.
+
+from ducktape.mark import matrix
+from ducktape.mark.resource import cluster
+from ducktape.utils.util import wait_until
+from kafkatest.tests.verifiable_share_consumer_test import VerifiableShareConsumerTest
+
+from kafkatest.services.kafka import TopicPartition, quorum
+
+import signal
+
+class ShareTest(VerifiableShareConsumerTest):
Review Comment:
class ShareConsmerTest(VerifiableShareConsumerTest)?
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,627 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
Review Comment:
Why do we want to extend the classes like this? What benefit we are getting?
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH
+
+class ConsumerState:
+ Started = 1
+ Dead = 2
+
+class ShareConsumerEventHandler(object):
+
+ def __init__(self, node, idx, state=ConsumerState.Dead):
+ self.node = node
+ self.idx = idx
+ self.total_consumed = 0
+ self.total_acknowledged = 0
+ self.total_acknowledged_failed = 0
+ self.consumed_per_partition = {}
+ self.acknowledged_per_partition = {}
+ self.acknowledged_per_partition_failed = {}
+ self.state = state
+
+ def handle_shutdown_complete(self, node=None, logger=None):
+ self.state = ConsumerState.Dead
+ if node is not None and logger is not None:
+ logger.debug("Shut down %s" % node.account.hostname)
+
+ def handle_startup_complete(self, node, logger):
+ self.state = ConsumerState.Started
+ logger.debug("Started %s" % node.account.hostname)
+
+ def handle_offsets_acknowledged(self, event, node, logger):
+ if event["success"]:
+ self.total_acknowledged += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition[topic_partition] = self.acknowledged_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
Review Comment:
How do we handle data when there is partial failure i.e. request is successful but some topic partitions acknowledgement failed?
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH
+
+class ConsumerState:
+ Started = 1
+ Dead = 2
+
+class ShareConsumerEventHandler(object):
+
+ def __init__(self, node, idx, state=ConsumerState.Dead):
+ self.node = node
+ self.idx = idx
+ self.total_consumed = 0
+ self.total_acknowledged = 0
+ self.total_acknowledged_failed = 0
+ self.consumed_per_partition = {}
+ self.acknowledged_per_partition = {}
+ self.acknowledged_per_partition_failed = {}
+ self.state = state
+
+ def handle_shutdown_complete(self, node=None, logger=None):
+ self.state = ConsumerState.Dead
+ if node is not None and logger is not None:
+ logger.debug("Shut down %s" % node.account.hostname)
+
+ def handle_startup_complete(self, node, logger):
+ self.state = ConsumerState.Started
+ logger.debug("Started %s" % node.account.hostname)
+
+ def handle_offsets_acknowledged(self, event, node, logger):
+ if event["success"]:
+ self.total_acknowledged += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition[topic_partition] = self.acknowledged_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ else:
+ self.total_acknowledged_failed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition_failed[topic_partition] = self.acknowledged_per_partition_failed.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ logger.debug("Offset acknowledgement failed for: %s" % (node.account.hostname))
+
+ def handle_records_consumed(self, event, node, logger):
+ self.total_consumed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.consumed_per_partition[topic_partition] = self.consumed_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets consumed for %s" % (node.account.hostname))
+
+
+ def handle_kill_process(self, clean_shutdown):
+ # if the shutdown was clean, then we expect the explicit
+ # shutdown event from the consumer
Review Comment:
nit:
```suggestion
# shutdown event from the share consumer
```
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,627 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
Review Comment:
Can't we use TopicPartition.java class?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] First draft of system tests for the new share consumer client [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1916356233
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH
+
+class ConsumerState:
+ Started = 1
+ Dead = 2
+
+class ShareConsumerEventHandler(object):
+
+ def __init__(self, node, idx, state=ConsumerState.Dead):
+ self.node = node
+ self.idx = idx
+ self.total_consumed = 0
+ self.total_acknowledged = 0
+ self.total_acknowledged_failed = 0
+ self.consumed_per_partition = {}
+ self.acknowledged_per_partition = {}
+ self.acknowledged_per_partition_failed = {}
+ self.state = state
+
+ def handle_shutdown_complete(self, node=None, logger=None):
+ self.state = ConsumerState.Dead
+ if node is not None and logger is not None:
+ logger.debug("Shut down %s" % node.account.hostname)
+
+ def handle_startup_complete(self, node, logger):
+ self.state = ConsumerState.Started
+ logger.debug("Started %s" % node.account.hostname)
+
+ def handle_offsets_acknowledged(self, event, node, logger):
+ if event["success"]:
+ self.total_acknowledged += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition[topic_partition] = self.acknowledged_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ else:
+ self.total_acknowledged_failed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition_failed[topic_partition] = self.acknowledged_per_partition_failed.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ logger.debug("Offset acknowledgement failed for: %s" % (node.account.hostname))
+
+ def handle_records_consumed(self, event, node, logger):
+ self.total_consumed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.consumed_per_partition[topic_partition] = self.consumed_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets consumed for %s" % (node.account.hostname))
+
+
+ def handle_kill_process(self, clean_shutdown):
+ # if the shutdown was clean, then we expect the explicit
+ # shutdown event from the consumer
+ if not clean_shutdown:
+ self.handle_shutdown_complete()
+
+class VerifiableShareConsumer(KafkaPathResolverMixin, VerifiableClientMixin, BackgroundThreadService):
+ """This service wraps org.apache.kafka.tools.VerifiableShareConsumer for use in
+ system testing.
+
+ NOTE: this class should be treated as a PUBLIC API. Downstream users use
+ this service both directly and through class extension, so care must be
+ taken to ensure compatibility.
+ """
+
+ PERSISTENT_ROOT = "/mnt/verifiable_share_consumer"
+ STDOUT_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stdout")
+ STDERR_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stderr")
+ LOG_DIR = os.path.join(PERSISTENT_ROOT, "logs")
+ LOG_FILE = os.path.join(LOG_DIR, "verifiable_share_consumer.log")
+ LOG4J_CONFIG = os.path.join(PERSISTENT_ROOT, "tools-log4j.properties")
+ CONFIG_FILE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.properties")
+
+ logs = {
+ "verifiable_share_consumer_stdout": {
+ "path": STDOUT_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_stderr": {
+ "path": STDERR_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_log": {
+ "path": LOG_FILE,
+ "collect_default": True}
+ }
+
+ def __init__(self, context, num_nodes, kafka, topic, group_id,
+ max_messages=-1, acknowledgement_mode="auto", offset_reset_strategy="",
+ version=DEV_BRANCH, stop_timeout_sec=60, log_level="INFO", jaas_override_variables=None,
+ on_record_consumed=None):
+ """
+ :param jaas_override_variables: A dict of variables to be used in the jaas.conf template file
+ """
+ super(VerifiableShareConsumer, self).__init__(context, num_nodes)
+ self.log_level = log_level
+ self.kafka = kafka
+ self.topic = topic
+ self.group_id = group_id
+ self.offset_reset_strategy = offset_reset_strategy
+ self.max_messages = max_messages
+ self.acknowledgement_mode = acknowledgement_mode
+ self.prop_file = ""
+ self.stop_timeout_sec = stop_timeout_sec
+ self.on_record_consumed = on_record_consumed
+
+ self.event_handlers = {}
+ self.jaas_override_variables = jaas_override_variables or {}
+
+ self.total_records_consumed = 0
+ self.total_records_acknowledged = 0
+ self.total_records_acknowledged_failed = 0
+ self.consumed_records_offsets = set()
+ # self.consumed_more_than_once = []
Review Comment:
This should probably be removed.
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,627 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
Review Comment:
I wonder whether a separate event might make more sense for the unsuccessful case. wdyt? Maybe an enhancement for later.
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,627 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
Review Comment:
`@JsonProperty` I think.
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH
+
+class ConsumerState:
Review Comment:
I'd go for `ShareConsumerState` I think.
##########
tests/kafkatest/tests/verifiable_share_consumer_test.py:
##########
@@ -0,0 +1,106 @@
+# 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.
+
+from ducktape.utils.util import wait_until
+
+from kafkatest.tests.kafka_test import KafkaTest
+from kafkatest.services.verifiable_producer import VerifiableProducer
+from kafkatest.services.verifiable_share_consumer import VerifiableShareConsumer
+from kafkatest.services.kafka import TopicPartition
+
+class VerifiableShareConsumerTest(KafkaTest):
+ PRODUCER_REQUEST_TIMEOUT_SEC = 30
+
+ def __init__(self, test_context, num_consumers=1, num_producers=0, **kwargs):
+ super(VerifiableShareConsumerTest, self).__init__(test_context, **kwargs)
+ self.num_consumers = num_consumers
+ self.num_producers = num_producers
+
+ def _all_partitions(self, topic, num_partitions):
+ partitions = set()
+ for i in range(num_partitions):
+ partitions.add(TopicPartition(topic=topic, partition=i))
+ return partitions
+
+ def _partitions(self, assignment):
+ partitions = []
+ for parts in assignment.values():
+ partitions += parts
+ return partitions
+
+ def valid_assignment(self, topic, num_partitions, assignment):
+ all_partitions = self._all_partitions(topic, num_partitions)
+ partitions = self._partitions(assignment)
+ return len(partitions) == num_partitions and set(partitions) == all_partitions
+
+ def min_cluster_size(self):
+ """Override this since we're adding services outside of the constructor"""
+ return super(VerifiableShareConsumerTest, self).min_cluster_size() + self.num_consumers + self.num_producers
+
+ def setup_share_group(self, topic, acknowledgement_mode="auto", group_id="test_group_id", offset_reset_strategy="", **kwargs):
+ return VerifiableShareConsumer(self.test_context, self.num_consumers, self.kafka,
+ topic, group_id, acknowledgement_mode=acknowledgement_mode,
+ offset_reset_strategy=offset_reset_strategy, log_level="TRACE", **kwargs)
+
+ def setup_producer(self, topic, max_messages=-1, throughput=500):
+ return VerifiableProducer(self.test_context, self.num_producers, self.kafka, topic,
+ max_messages=max_messages, throughput=throughput,
+ request_timeout_sec=self.PRODUCER_REQUEST_TIMEOUT_SEC,
+ log_level="DEBUG")
+
+ def await_produced_messages(self, producer, min_messages=1000, timeout_sec=10):
+ current_acked = producer.num_acked
+ wait_until(lambda: producer.num_acked >= current_acked + min_messages, timeout_sec=timeout_sec,
+ err_msg="Timeout awaiting messages to be produced and acked")
+
+ def await_consumed_messages(self, consumer, min_messages=1, timeout_sec=10, total=False):
+ current_total = 0
+ if total is False:
+ current_total = consumer.total_consumed()
+ wait_until(lambda: consumer.total_consumed() >= current_total + min_messages,
+ timeout_sec=timeout_sec,
+ err_msg="Timed out waiting for consumption")
+
+ def await_consumed_messages_by_a_consumer(self, consumer, node, min_messages=1, timeout_sec=10, total=False):
Review Comment:
Can you explain the purpose of `await_consumed_messages_by_a_consumer` as opposed to `await_consumed_messages`? This is done in a different way than the regular `consumer_test.py` and `verifiable_consumer_test.py` and I wonder what the motivation is for doing it differently.
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH
+
+class ConsumerState:
+ Started = 1
+ Dead = 2
+
+class ShareConsumerEventHandler(object):
+
+ def __init__(self, node, idx, state=ConsumerState.Dead):
+ self.node = node
+ self.idx = idx
+ self.total_consumed = 0
+ self.total_acknowledged = 0
+ self.total_acknowledged_failed = 0
+ self.consumed_per_partition = {}
+ self.acknowledged_per_partition = {}
+ self.acknowledged_per_partition_failed = {}
+ self.state = state
+
+ def handle_shutdown_complete(self, node=None, logger=None):
+ self.state = ConsumerState.Dead
+ if node is not None and logger is not None:
+ logger.debug("Shut down %s" % node.account.hostname)
+
+ def handle_startup_complete(self, node, logger):
+ self.state = ConsumerState.Started
+ logger.debug("Started %s" % node.account.hostname)
+
+ def handle_offsets_acknowledged(self, event, node, logger):
+ if event["success"]:
+ self.total_acknowledged += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition[topic_partition] = self.acknowledged_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ else:
+ self.total_acknowledged_failed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition_failed[topic_partition] = self.acknowledged_per_partition_failed.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ logger.debug("Offset acknowledgement failed for: %s" % (node.account.hostname))
+
+ def handle_records_consumed(self, event, node, logger):
+ self.total_consumed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.consumed_per_partition[topic_partition] = self.consumed_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets consumed for %s" % (node.account.hostname))
+
+
+ def handle_kill_process(self, clean_shutdown):
+ # if the shutdown was clean, then we expect the explicit
+ # shutdown event from the consumer
+ if not clean_shutdown:
+ self.handle_shutdown_complete()
+
+class VerifiableShareConsumer(KafkaPathResolverMixin, VerifiableClientMixin, BackgroundThreadService):
+ """This service wraps org.apache.kafka.tools.VerifiableShareConsumer for use in
+ system testing.
+
+ NOTE: this class should be treated as a PUBLIC API. Downstream users use
+ this service both directly and through class extension, so care must be
+ taken to ensure compatibility.
+ """
+
+ PERSISTENT_ROOT = "/mnt/verifiable_share_consumer"
+ STDOUT_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stdout")
+ STDERR_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stderr")
+ LOG_DIR = os.path.join(PERSISTENT_ROOT, "logs")
+ LOG_FILE = os.path.join(LOG_DIR, "verifiable_share_consumer.log")
+ LOG4J_CONFIG = os.path.join(PERSISTENT_ROOT, "tools-log4j.properties")
+ CONFIG_FILE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.properties")
+
+ logs = {
+ "verifiable_share_consumer_stdout": {
+ "path": STDOUT_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_stderr": {
+ "path": STDERR_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_log": {
+ "path": LOG_FILE,
+ "collect_default": True}
+ }
+
+ def __init__(self, context, num_nodes, kafka, topic, group_id,
+ max_messages=-1, acknowledgement_mode="auto", offset_reset_strategy="",
+ version=DEV_BRANCH, stop_timeout_sec=60, log_level="INFO", jaas_override_variables=None,
+ on_record_consumed=None):
+ """
+ :param jaas_override_variables: A dict of variables to be used in the jaas.conf template file
+ """
+ super(VerifiableShareConsumer, self).__init__(context, num_nodes)
+ self.log_level = log_level
+ self.kafka = kafka
+ self.topic = topic
+ self.group_id = group_id
+ self.offset_reset_strategy = offset_reset_strategy
+ self.max_messages = max_messages
+ self.acknowledgement_mode = acknowledgement_mode
+ self.prop_file = ""
+ self.stop_timeout_sec = stop_timeout_sec
+ self.on_record_consumed = on_record_consumed
+
+ self.event_handlers = {}
+ self.jaas_override_variables = jaas_override_variables or {}
+
+ self.total_records_consumed = 0
+ self.total_records_acknowledged = 0
+ self.total_records_acknowledged_failed = 0
+ self.consumed_records_offsets = set()
+ # self.consumed_more_than_once = []
+
+ self.acknowledged_records_offsets = set()
+ # self.acknowledged_more_than_once = []
Review Comment:
And this.
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH
+
+class ConsumerState:
+ Started = 1
+ Dead = 2
+
+class ShareConsumerEventHandler(object):
+
+ def __init__(self, node, idx, state=ConsumerState.Dead):
+ self.node = node
+ self.idx = idx
+ self.total_consumed = 0
+ self.total_acknowledged = 0
+ self.total_acknowledged_failed = 0
+ self.consumed_per_partition = {}
+ self.acknowledged_per_partition = {}
+ self.acknowledged_per_partition_failed = {}
+ self.state = state
+
+ def handle_shutdown_complete(self, node=None, logger=None):
+ self.state = ConsumerState.Dead
+ if node is not None and logger is not None:
+ logger.debug("Shut down %s" % node.account.hostname)
+
+ def handle_startup_complete(self, node, logger):
+ self.state = ConsumerState.Started
+ logger.debug("Started %s" % node.account.hostname)
+
+ def handle_offsets_acknowledged(self, event, node, logger):
+ if event["success"]:
+ self.total_acknowledged += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition[topic_partition] = self.acknowledged_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ else:
+ self.total_acknowledged_failed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition_failed[topic_partition] = self.acknowledged_per_partition_failed.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ logger.debug("Offset acknowledgement failed for: %s" % (node.account.hostname))
+
+ def handle_records_consumed(self, event, node, logger):
+ self.total_consumed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.consumed_per_partition[topic_partition] = self.consumed_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets consumed for %s" % (node.account.hostname))
+
+
+ def handle_kill_process(self, clean_shutdown):
+ # if the shutdown was clean, then we expect the explicit
+ # shutdown event from the consumer
+ if not clean_shutdown:
+ self.handle_shutdown_complete()
+
+class VerifiableShareConsumer(KafkaPathResolverMixin, VerifiableClientMixin, BackgroundThreadService):
+ """This service wraps org.apache.kafka.tools.VerifiableShareConsumer for use in
+ system testing.
+
+ NOTE: this class should be treated as a PUBLIC API. Downstream users use
+ this service both directly and through class extension, so care must be
+ taken to ensure compatibility.
+ """
+
+ PERSISTENT_ROOT = "/mnt/verifiable_share_consumer"
+ STDOUT_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stdout")
+ STDERR_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stderr")
+ LOG_DIR = os.path.join(PERSISTENT_ROOT, "logs")
+ LOG_FILE = os.path.join(LOG_DIR, "verifiable_share_consumer.log")
+ LOG4J_CONFIG = os.path.join(PERSISTENT_ROOT, "tools-log4j.properties")
+ CONFIG_FILE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.properties")
+
+ logs = {
+ "verifiable_share_consumer_stdout": {
+ "path": STDOUT_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_stderr": {
+ "path": STDERR_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_log": {
+ "path": LOG_FILE,
+ "collect_default": True}
+ }
+
+ def __init__(self, context, num_nodes, kafka, topic, group_id,
+ max_messages=-1, acknowledgement_mode="auto", offset_reset_strategy="",
+ version=DEV_BRANCH, stop_timeout_sec=60, log_level="INFO", jaas_override_variables=None,
+ on_record_consumed=None):
+ """
+ :param jaas_override_variables: A dict of variables to be used in the jaas.conf template file
+ """
+ super(VerifiableShareConsumer, self).__init__(context, num_nodes)
+ self.log_level = log_level
+ self.kafka = kafka
+ self.topic = topic
+ self.group_id = group_id
+ self.offset_reset_strategy = offset_reset_strategy
+ self.max_messages = max_messages
+ self.acknowledgement_mode = acknowledgement_mode
+ self.prop_file = ""
+ self.stop_timeout_sec = stop_timeout_sec
+ self.on_record_consumed = on_record_consumed
+
+ self.event_handlers = {}
+ self.jaas_override_variables = jaas_override_variables or {}
+
+ self.total_records_consumed = 0
+ self.total_records_acknowledged = 0
+ self.total_records_acknowledged_failed = 0
+ self.consumed_records_offsets = set()
+ # self.consumed_more_than_once = []
+
+ self.acknowledged_records_offsets = set()
+ # self.acknowledged_more_than_once = []
+ self.is_offset_reset_strategy_set = False
+
+ for node in self.nodes:
+ node.version = version
+
+ def java_class_name(self):
+ return "VerifiableShareConsumer"
+
+ def create_event_handler(self, idx, node):
+ return ShareConsumerEventHandler(node, idx)
+
+ def _worker(self, idx, node):
+ with self.lock:
+ self.event_handlers[node] = self.create_event_handler(idx, node)
+ handler = self.event_handlers[node]
+
+ node.account.ssh("mkdir -p %s" % VerifiableShareConsumer.PERSISTENT_ROOT, allow_fail=False)
+
+ # Create and upload log properties
+ log_config = self.render('tools_log4j.properties', log_file=VerifiableShareConsumer.LOG_FILE)
Review Comment:
Please investigate `get_log4j_config_for_tools` in the other related Python code. I think this way of doing things has been superseded.
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH
+
+class ConsumerState:
+ Started = 1
+ Dead = 2
+
+class ShareConsumerEventHandler(object):
+
+ def __init__(self, node, idx, state=ConsumerState.Dead):
+ self.node = node
+ self.idx = idx
+ self.total_consumed = 0
+ self.total_acknowledged = 0
+ self.total_acknowledged_failed = 0
+ self.consumed_per_partition = {}
+ self.acknowledged_per_partition = {}
+ self.acknowledged_per_partition_failed = {}
+ self.state = state
+
+ def handle_shutdown_complete(self, node=None, logger=None):
+ self.state = ConsumerState.Dead
+ if node is not None and logger is not None:
+ logger.debug("Shut down %s" % node.account.hostname)
+
+ def handle_startup_complete(self, node, logger):
+ self.state = ConsumerState.Started
+ logger.debug("Started %s" % node.account.hostname)
+
+ def handle_offsets_acknowledged(self, event, node, logger):
+ if event["success"]:
+ self.total_acknowledged += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition[topic_partition] = self.acknowledged_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ else:
+ self.total_acknowledged_failed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition_failed[topic_partition] = self.acknowledged_per_partition_failed.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ logger.debug("Offset acknowledgement failed for: %s" % (node.account.hostname))
+
+ def handle_records_consumed(self, event, node, logger):
+ self.total_consumed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.consumed_per_partition[topic_partition] = self.consumed_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets consumed for %s" % (node.account.hostname))
+
+
+ def handle_kill_process(self, clean_shutdown):
+ # if the shutdown was clean, then we expect the explicit
+ # shutdown event from the consumer
+ if not clean_shutdown:
+ self.handle_shutdown_complete()
+
+class VerifiableShareConsumer(KafkaPathResolverMixin, VerifiableClientMixin, BackgroundThreadService):
+ """This service wraps org.apache.kafka.tools.VerifiableShareConsumer for use in
+ system testing.
+
+ NOTE: this class should be treated as a PUBLIC API. Downstream users use
+ this service both directly and through class extension, so care must be
+ taken to ensure compatibility.
+ """
+
+ PERSISTENT_ROOT = "/mnt/verifiable_share_consumer"
+ STDOUT_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stdout")
+ STDERR_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stderr")
+ LOG_DIR = os.path.join(PERSISTENT_ROOT, "logs")
+ LOG_FILE = os.path.join(LOG_DIR, "verifiable_share_consumer.log")
+ LOG4J_CONFIG = os.path.join(PERSISTENT_ROOT, "tools-log4j.properties")
+ CONFIG_FILE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.properties")
+
+ logs = {
+ "verifiable_share_consumer_stdout": {
+ "path": STDOUT_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_stderr": {
+ "path": STDERR_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_log": {
+ "path": LOG_FILE,
+ "collect_default": True}
+ }
+
+ def __init__(self, context, num_nodes, kafka, topic, group_id,
+ max_messages=-1, acknowledgement_mode="auto", offset_reset_strategy="",
+ version=DEV_BRANCH, stop_timeout_sec=60, log_level="INFO", jaas_override_variables=None,
+ on_record_consumed=None):
+ """
+ :param jaas_override_variables: A dict of variables to be used in the jaas.conf template file
+ """
+ super(VerifiableShareConsumer, self).__init__(context, num_nodes)
+ self.log_level = log_level
+ self.kafka = kafka
+ self.topic = topic
+ self.group_id = group_id
+ self.offset_reset_strategy = offset_reset_strategy
+ self.max_messages = max_messages
+ self.acknowledgement_mode = acknowledgement_mode
+ self.prop_file = ""
+ self.stop_timeout_sec = stop_timeout_sec
+ self.on_record_consumed = on_record_consumed
+
+ self.event_handlers = {}
+ self.jaas_override_variables = jaas_override_variables or {}
+
+ self.total_records_consumed = 0
+ self.total_records_acknowledged = 0
+ self.total_records_acknowledged_failed = 0
+ self.consumed_records_offsets = set()
+ # self.consumed_more_than_once = []
+
+ self.acknowledged_records_offsets = set()
+ # self.acknowledged_more_than_once = []
+ self.is_offset_reset_strategy_set = False
+
+ for node in self.nodes:
+ node.version = version
+
+ def java_class_name(self):
+ return "VerifiableShareConsumer"
+
+ def create_event_handler(self, idx, node):
+ return ShareConsumerEventHandler(node, idx)
+
+ def _worker(self, idx, node):
+ with self.lock:
+ self.event_handlers[node] = self.create_event_handler(idx, node)
+ handler = self.event_handlers[node]
+
+ node.account.ssh("mkdir -p %s" % VerifiableShareConsumer.PERSISTENT_ROOT, allow_fail=False)
+
+ # Create and upload log properties
+ log_config = self.render('tools_log4j.properties', log_file=VerifiableShareConsumer.LOG_FILE)
+ node.account.create_file(VerifiableShareConsumer.LOG4J_CONFIG, log_config)
+
+ # Create and upload config file
+ self.security_config = self.kafka.security_config.client_config(self.prop_file, node,
+ self.jaas_override_variables)
+ self.security_config.setup_node(node)
+ self.prop_file += str(self.security_config)
+ self.logger.info("verifiable_share_consumer.properties:")
+ self.logger.info(self.prop_file)
+ node.account.create_file(VerifiableShareConsumer.CONFIG_FILE, self.prop_file)
+ self.security_config.setup_node(node)
+
+ cmd = self.start_cmd(node)
+ self.logger.debug("VerifiableShareConsumer %d command: %s" % (idx, cmd))
+
+ for line in node.account.ssh_capture(cmd):
+ event = self.try_parse_json(node, line.strip())
+ if event is not None:
+ with self.lock:
+ name = event["name"]
+ if name == "shutdown_complete":
+ handler.handle_shutdown_complete(node, self.logger)
+ elif name == "startup_complete":
+ handler.handle_startup_complete(node, self.logger)
+ elif name == "offsets_acknowledged":
+ handler.handle_offsets_acknowledged(event, node, self.logger)
+ self._update_global_acknowledged(event)
+ elif name == "records_consumed":
+ handler.handle_records_consumed(event, node, self.logger)
+ self._update_global_consumed(event)
+ elif name == "record_data" and self.on_record_consumed:
+ self.on_record_consumed(event, node)
+ elif name == "offset_reset_strategy_set":
+ self._on_offset_reset_strategy_set()
+ else:
+ self.logger.debug("%s: ignoring unknown event: %s" % (str(node.account), event))
+
+ def _update_global_acknowledged(self, acknowledge_event):
+ if acknowledge_event["success"]:
+ self.total_records_acknowledged += acknowledge_event["count"]
+ else:
+ self.total_records_acknowledged_failed += acknowledge_event["count"]
+ for share_partition_data in acknowledge_event["partitions"]:
+ tpkey = str(share_partition_data["topic"]) + "-" + str(share_partition_data["partition"])
+ for offset in share_partition_data["offsets"]:
+ key = tpkey + "-" + str(offset)
+ if key not in self.acknowledged_records_offsets:
+ self.acknowledged_records_offsets.add(key)
+
+ def _update_global_consumed(self, consumed_event):
+ self.total_records_consumed += consumed_event["count"]
+
+ for share_partition_data in consumed_event["partitions"]:
+ tpkey = str(share_partition_data["topic"]) + "-" + str(share_partition_data["partition"])
+ for offset in share_partition_data["offsets"]:
+ key = tpkey + "-" + str(offset)
+ if key not in self.consumed_records_offsets:
+ self.consumed_records_offsets.add(key)
+
+ def _on_offset_reset_strategy_set(self):
+ self.is_offset_reset_strategy_set = True
+
+ def start_cmd(self, node):
+ cmd = ""
+ cmd += "export LOG_DIR=%s;" % VerifiableShareConsumer.LOG_DIR
+ cmd += " export KAFKA_OPTS=%s;" % self.security_config.kafka_opts
+ cmd += " export KAFKA_LOG4J_OPTS=\"-Dlog4j.configuration=file:%s\"; " % VerifiableShareConsumer.LOG4J_CONFIG
+ cmd += self.impl.exec_cmd(node)
+ if self.on_record_consumed:
+ cmd += " --verbose"
+
+ # 3.7.0 includes support for KIP-848 which introduced a new implementation of the consumer group protocol.
Review Comment:
These lines about 3.7.0 are not relevant in the verifiable_share_consumer.
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,627 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
Review Comment:
Probably ought to have a `@JsonPropertyOrder` too.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Infrastructure for system tests for the new share consumer client [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1919928924
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,627 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
Review Comment:
OK. I expect we will need to do that. Happy to defer that until we have increased the sophistication of the tests. This is going to get tricky :)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Infrastructure for system tests for the new share consumer client [kafka]
Posted by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org>.
chirag-wadhwa5 commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1918778954
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,627 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
Review Comment:
Thanks for review. Currently, the tests aren't intelligent enough to analyze the exceptions returned and take actions regarding that. The tests simply check for whether the acknowledgement was successful or not and keep track of number of records successfully acknowledged or for which acknowledgement failed. If in future any test requires the analysis of the acknowledgement failure due to which the processing becomes a little complicated, then we can create a new failure event for that.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] First draft of system tests for the new share consumer client [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1904478263
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,626 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count = count;
+ this.partitions = partitions;
+ this.error = error;
+ this.success = success;
+ }
+
+ @Override
+ public String name() {
+ return "offsets_acknowledged";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<AcknowledgedData> partitions() {
+ return partitions;
+ }
+
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String error() {
+ return error;
+ }
+
+ @JsonProperty
+ public boolean success() {
+ return success;
+ }
+
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" })
+ public static class RecordData extends ShareConsumerEvent {
+
+ private final ConsumerRecord<String, String> record;
+
+ public RecordData(ConsumerRecord<String, String> record) {
+ this.record = record;
+ }
+
+ @Override
+ public String name() {
+ return "record_data";
+ }
+
+ @JsonProperty
+ public String topic() {
+ return record.topic();
+ }
+
+ @JsonProperty
+ public int partition() {
+ return record.partition();
+ }
+
+ @JsonProperty
+ public String key() {
+ return record.key();
+ }
+
+ @JsonProperty
+ public String value() {
+ return record.value();
+ }
+
+ @JsonProperty
+ public long offset() {
+ return record.offset();
+ }
+
+ }
+
+ public VerifiableShareConsumer(KafkaShareConsumer<String, String> consumer,
+ PrintStream out,
+ Integer maxMessages,
+ String topic,
+ AcknowledgementMode acknowledgementMode,
+ String offsetResetStrategy,
+ String brokerHostandPort,
+ String groupId,
+ Boolean verbose) {
+ this.out = out;
+ this.consumer = consumer;
+ this.topic = topic;
+ this.acknowledgementMode = acknowledgementMode;
+ this.offsetResetStrategy = offsetResetStrategy;
+ this.verbose = verbose;
+ this.maxMessages = maxMessages;
+ this.brokerHostandPort = brokerHostandPort;
+ this.groupId = groupId;
+ addKafkaSerializerModule();
+ }
+
+ private void addKafkaSerializerModule() {
+ SimpleModule kafka = new SimpleModule();
+ kafka.addSerializer(TopicPartition.class, new JsonSerializer<>() {
+ @Override
+ public void serialize(TopicPartition tp, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+ gen.writeStartObject();
+ gen.writeObjectField("topic", tp.topic());
+ gen.writeObjectField("partition", tp.partition());
+ gen.writeEndObject();
+ }
+ });
+ mapper.registerModule(kafka);
+ }
+
+ private void onRecordsReceived(ConsumerRecords<String, String> records) {
+
Review Comment:
nit: Extra blank line
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,626 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count = count;
+ this.partitions = partitions;
+ this.error = error;
+ this.success = success;
+ }
+
+ @Override
+ public String name() {
+ return "offsets_acknowledged";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<AcknowledgedData> partitions() {
+ return partitions;
+ }
+
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String error() {
+ return error;
+ }
+
+ @JsonProperty
+ public boolean success() {
+ return success;
+ }
+
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" })
+ public static class RecordData extends ShareConsumerEvent {
+
+ private final ConsumerRecord<String, String> record;
+
+ public RecordData(ConsumerRecord<String, String> record) {
+ this.record = record;
+ }
+
+ @Override
+ public String name() {
+ return "record_data";
+ }
+
+ @JsonProperty
+ public String topic() {
+ return record.topic();
+ }
+
+ @JsonProperty
+ public int partition() {
+ return record.partition();
+ }
+
+ @JsonProperty
+ public String key() {
+ return record.key();
+ }
+
+ @JsonProperty
+ public String value() {
+ return record.value();
+ }
+
+ @JsonProperty
+ public long offset() {
+ return record.offset();
+ }
+
+ }
+
+ public VerifiableShareConsumer(KafkaShareConsumer<String, String> consumer,
+ PrintStream out,
+ Integer maxMessages,
+ String topic,
+ AcknowledgementMode acknowledgementMode,
+ String offsetResetStrategy,
+ String brokerHostandPort,
+ String groupId,
+ Boolean verbose) {
+ this.out = out;
+ this.consumer = consumer;
+ this.topic = topic;
+ this.acknowledgementMode = acknowledgementMode;
+ this.offsetResetStrategy = offsetResetStrategy;
+ this.verbose = verbose;
+ this.maxMessages = maxMessages;
+ this.brokerHostandPort = brokerHostandPort;
+ this.groupId = groupId;
+ addKafkaSerializerModule();
+ }
+
+ private void addKafkaSerializerModule() {
+ SimpleModule kafka = new SimpleModule();
+ kafka.addSerializer(TopicPartition.class, new JsonSerializer<>() {
+ @Override
+ public void serialize(TopicPartition tp, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+ gen.writeStartObject();
+ gen.writeObjectField("topic", tp.topic());
+ gen.writeObjectField("partition", tp.partition());
+ gen.writeEndObject();
+ }
+ });
+ mapper.registerModule(kafka);
+ }
+
+ private void onRecordsReceived(ConsumerRecords<String, String> records) {
+
+ List<RecordSetSummary> summaries = new ArrayList<>();
+ for (TopicPartition tp : records.partitions()) {
+ List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
+
+ if (partitionRecords.isEmpty())
+ continue;
+
+ Set<Long> partitionOffsets = new HashSet<>();
Review Comment:
This would be better as a sorted set. I suggest using TreeSet.
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,626 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count = count;
+ this.partitions = partitions;
+ this.error = error;
+ this.success = success;
+ }
+
+ @Override
+ public String name() {
+ return "offsets_acknowledged";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<AcknowledgedData> partitions() {
+ return partitions;
+ }
+
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String error() {
+ return error;
+ }
+
+ @JsonProperty
+ public boolean success() {
+ return success;
+ }
+
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" })
+ public static class RecordData extends ShareConsumerEvent {
+
+ private final ConsumerRecord<String, String> record;
+
+ public RecordData(ConsumerRecord<String, String> record) {
+ this.record = record;
+ }
+
+ @Override
+ public String name() {
+ return "record_data";
+ }
+
+ @JsonProperty
+ public String topic() {
+ return record.topic();
+ }
+
+ @JsonProperty
+ public int partition() {
+ return record.partition();
+ }
+
+ @JsonProperty
+ public String key() {
+ return record.key();
+ }
+
+ @JsonProperty
+ public String value() {
+ return record.value();
+ }
+
+ @JsonProperty
+ public long offset() {
+ return record.offset();
+ }
+
+ }
+
+ public VerifiableShareConsumer(KafkaShareConsumer<String, String> consumer,
+ PrintStream out,
+ Integer maxMessages,
+ String topic,
+ AcknowledgementMode acknowledgementMode,
+ String offsetResetStrategy,
+ String brokerHostandPort,
+ String groupId,
+ Boolean verbose) {
+ this.out = out;
+ this.consumer = consumer;
+ this.topic = topic;
+ this.acknowledgementMode = acknowledgementMode;
+ this.offsetResetStrategy = offsetResetStrategy;
+ this.verbose = verbose;
+ this.maxMessages = maxMessages;
+ this.brokerHostandPort = brokerHostandPort;
+ this.groupId = groupId;
+ addKafkaSerializerModule();
+ }
+
+ private void addKafkaSerializerModule() {
+ SimpleModule kafka = new SimpleModule();
+ kafka.addSerializer(TopicPartition.class, new JsonSerializer<>() {
+ @Override
+ public void serialize(TopicPartition tp, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+ gen.writeStartObject();
+ gen.writeObjectField("topic", tp.topic());
+ gen.writeObjectField("partition", tp.partition());
+ gen.writeEndObject();
+ }
+ });
+ mapper.registerModule(kafka);
+ }
+
+ private void onRecordsReceived(ConsumerRecords<String, String> records) {
+
+ List<RecordSetSummary> summaries = new ArrayList<>();
+ for (TopicPartition tp : records.partitions()) {
+ List<ConsumerRecord<String, String>> partitionRecords = records.records(tp);
+
+ if (partitionRecords.isEmpty())
+ continue;
+
+ Set<Long> partitionOffsets = new HashSet<>();
+
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ partitionOffsets.add(record.offset());
+ }
+
+ summaries.add(new RecordSetSummary(tp.topic(), tp.partition(), partitionOffsets));
+
+ if (this.verbose) {
+ for (ConsumerRecord<String, String> record : partitionRecords) {
+ printJson(new RecordData(record));
+ }
+ }
+ }
+
+ printJson(new RecordsConsumed(records.count(), summaries));
+ }
+
+ @Override
+ public void onComplete(Map<TopicIdPartition, Set<Long>> offsetsMap, Exception exception) {
+ List<AcknowledgedData> acknowledgedOffsets = new ArrayList<>();
+ int totalAcknowledged = 0;
+ for (Map.Entry<TopicIdPartition, Set<Long>> offsetEntry : offsetsMap.entrySet()) {
+ TopicIdPartition tp = offsetEntry.getKey();
+ acknowledgedOffsets.add(new AcknowledgedData(tp.topic(), tp.partition(), offsetEntry.getValue()));
+ totalAcknowledged += offsetEntry.getValue().size();
+ }
+
+ boolean success = true;
+ String error = null;
+ if (exception != null) {
+ success = false;
+ error = exception.getMessage();
+ }
+ printJson(new OffsetsAcknowledged(totalAcknowledged, acknowledgedOffsets, error, success));
+ if (success) {
+ this.totalAcknowledged += totalAcknowledged;
+ }
+ }
+
+ public void run() {
+ try {
+ printJson(new StartupComplete());
+
+ if (!Objects.equals(offsetResetStrategy, "")) {
+ ShareGroupAutoOffsetResetStrategy offsetResetStrategy =
+ ShareGroupAutoOffsetResetStrategy.fromString(this.offsetResetStrategy);
+
+ Properties adminClientProps = new Properties();
+ adminClientProps.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerHostandPort);
+
+ Admin adminClient = Admin.create(adminClientProps);
+
+ ConfigResource configResource = new ConfigResource(ConfigResource.Type.GROUP, groupId);
+ Map<ConfigResource, Collection<AlterConfigOp>> alterEntries = new HashMap<>();
+ alterEntries.put(configResource, List.of(new AlterConfigOp(new ConfigEntry(
+ GroupConfig.SHARE_AUTO_OFFSET_RESET_CONFIG, offsetResetStrategy.type().toString()), AlterConfigOp.OpType.SET)));
+ AlterConfigsOptions alterOptions = new AlterConfigsOptions();
+
+ // Setting the share group auto offset reset strategy
+ adminClient.incrementalAlterConfigs(alterEntries, alterOptions)
+ .all()
+ .get(60, TimeUnit.SECONDS);
+
+ printJson(new OffsetResetStrategySet(offsetResetStrategy.type().toString()));
+ }
+
+ consumer.subscribe(Collections.singleton(this.topic));
+ consumer.setAcknowledgementCommitCallback(this);
+ while (!(maxMessages >= 0 && totalAcknowledged >= maxMessages)) {
+ ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5000));
+ if (!records.isEmpty()) {
+ onRecordsReceived(records);
+
+ if (acknowledgementMode == AcknowledgementMode.ASYNC) {
+ consumer.commitAsync();
+ } else if (acknowledgementMode == AcknowledgementMode.SYNC) {
+ Map<TopicIdPartition, Optional<KafkaException>> result = consumer.commitSync();
+ for (Map.Entry<TopicIdPartition, Optional<KafkaException>> resultEntry : result.entrySet()) {
+ if (resultEntry.getValue().isPresent()) {
+ log.error("Failed to commit offset synchronously for topic partition: {}", resultEntry.getKey());
+ }
+ }
+ }
+ }
+ }
+ } catch (WakeupException e) {
+ // ignore, we are closing
+ log.trace("Caught WakeupException because share consumer is shutdown, ignore and terminate.", e);
+ } catch (Throwable t) {
+ // Log the error, so it goes to the service log and not stdout
+ log.error("Error during processing, terminating share consumer process: ", t);
+ } finally {
+ consumer.close();
+ printJson(new ShutdownComplete());
+ shutdownLatch.countDown();
+ }
+ }
+
+ public void close() {
+ boolean interrupted = false;
+ try {
+ consumer.wakeup();
+ while (true) {
+ try {
+ shutdownLatch.await();
+ return;
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ }
+ } finally {
+ if (interrupted)
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ protected synchronized void printJson(Object data) {
+ try {
+ out.println(mapper.writeValueAsString(data));
+ } catch (JsonProcessingException e) {
+ out.println("Bad data can't be written as json: " + e.getMessage());
+ }
+ }
+
+ public enum AcknowledgementMode {
+ AUTO, ASYNC, SYNC;
+
+ @Override
+ public String toString() {
+ return super.toString().toLowerCase(Locale.ROOT);
+ }
+ }
+
+ private static ArgumentParser argParser() {
+ ArgumentParser parser = ArgumentParsers
+ .newArgumentParser("verifiable-share-consumer")
+ .defaultHelp(true)
+ .description("This tool creates a share group and consumes messages from a specific topic and emits share consumer events (e.g. share consumer startup, received messages, and offsets acknowledged) as JSON objects to STDOUT.");
+ MutuallyExclusiveGroup connectionGroup = parser.addMutuallyExclusiveGroup("Connection Group")
+ .description("Group of arguments for connection to brokers")
+ .required(true);
+ connectionGroup.addArgument("--bootstrap-server")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .metavar("HOST1:PORT1[,HOST2:PORT2[...]]")
+ .dest("bootstrapServer")
+ .help("The server(s) to connect to. Comma-separated list of Kafka brokers in the form HOST1:PORT1,HOST2:PORT2,...");
+
+ parser.addArgument("--topic")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .metavar("TOPIC")
+ .help("Consumes messages from this topic.");
+
+ parser.addArgument("--group-id")
+ .action(store())
+ .required(true)
+ .type(String.class)
+ .metavar("GROUP_ID")
+ .dest("groupId")
+ .help("The groupId shared among members of the consumer group");
Review Comment:
This is a "share group", not a "consumer group".
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Infrastructure for system tests for the new share consumer client [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1920091543
##########
tools/src/main/java/org/apache/kafka/tools/VerifiableShareConsumer.java:
##########
@@ -0,0 +1,630 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.tools;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.AlterConfigsOptions;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.consumer.AcknowledgementCommitCallback;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicIdPartition;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupConfig;
+import org.apache.kafka.coordinator.group.ShareGroupAutoOffsetResetStrategy;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+
+import net.sourceforge.argparse4j.ArgumentParsers;
+import net.sourceforge.argparse4j.inf.ArgumentParser;
+import net.sourceforge.argparse4j.inf.ArgumentParserException;
+import net.sourceforge.argparse4j.inf.MutuallyExclusiveGroup;
+import net.sourceforge.argparse4j.inf.Namespace;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static net.sourceforge.argparse4j.impl.Arguments.store;
+import static net.sourceforge.argparse4j.impl.Arguments.storeTrue;
+
+public class VerifiableShareConsumer implements Closeable, AcknowledgementCommitCallback {
+
+ private static final Logger log = LoggerFactory.getLogger(VerifiableShareConsumer.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final PrintStream out;
+ private final KafkaShareConsumer<String, String> consumer;
+ private final String topic;
+ private final AcknowledgementMode acknowledgementMode;
+ private final String offsetResetStrategy;
+ private final Boolean verbose;
+ private final int maxMessages;
+ private Integer totalAcknowledged = 0;
+ private final String brokerHostandPort;
+ private final String groupId;
+ private final CountDownLatch shutdownLatch = new CountDownLatch(1);
+
+ public static class PartitionData {
+ private final String topic;
+ private final int partition;
+
+ public PartitionData(String topic, int partition) {
+ this.topic = topic;
+ this.partition = partition;
+ }
+
+ @JsonProperty
+ public String topic() {
+ return topic;
+ }
+
+ @JsonProperty
+ public int partition() {
+ return partition;
+ }
+ }
+
+ public static class RecordSetSummary extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public RecordSetSummary(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+
+ }
+
+ protected static class AcknowledgedData extends PartitionData {
+ private final long count;
+ private final Set<Long> offsets;
+
+ public AcknowledgedData(String topic, int partition, Set<Long> offsets) {
+ super(topic, partition);
+ this.offsets = offsets;
+ this.count = offsets.size();
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public Set<Long> offsets() {
+ return offsets;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name" })
+ private abstract static class ShareConsumerEvent {
+ private final long timestamp = System.currentTimeMillis();
+
+ @JsonProperty
+ public abstract String name();
+
+ @JsonProperty
+ public long timestamp() {
+ return timestamp;
+ }
+ }
+
+ protected static class StartupComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "startup_complete";
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "offsetResetStrategy" })
+ protected static class OffsetResetStrategySet extends ShareConsumerEvent {
+
+ private final String offsetResetStrategy;
+
+ public OffsetResetStrategySet(String offsetResetStrategy) {
+ this.offsetResetStrategy = offsetResetStrategy;
+ }
+
+ @Override
+ public String name() {
+ return "offset_reset_strategy_set";
+ }
+
+ @JsonProperty
+ public String offsetResetStrategy() {
+ return offsetResetStrategy;
+ }
+ }
+
+ protected static class ShutdownComplete extends ShareConsumerEvent {
+
+ @Override
+ public String name() {
+ return "shutdown_complete";
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "count", "partitions" })
+ public static class RecordsConsumed extends ShareConsumerEvent {
+ private final long count;
+ private final List<RecordSetSummary> partitionSummaries;
+
+ public RecordsConsumed(long count, List<RecordSetSummary> partitionSummaries) {
+ this.count = count;
+ this.partitionSummaries = partitionSummaries;
+ }
+
+ @Override
+ public String name() {
+ return "records_consumed";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<RecordSetSummary> partitions() {
+ return partitionSummaries;
+ }
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "count", "partitions", "success", "error" })
+ protected static class OffsetsAcknowledged extends ShareConsumerEvent {
+
+ private final long count;
+ private final List<AcknowledgedData> partitions;
+ private final String error;
+ private final boolean success;
+
+ public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) {
+ this.count = count;
+ this.partitions = partitions;
+ this.error = error;
+ this.success = success;
+ }
+
+ @Override
+ public String name() {
+ return "offsets_acknowledged";
+ }
+
+ @JsonProperty
+ public long count() {
+ return count;
+ }
+
+ @JsonProperty
+ public List<AcknowledgedData> partitions() {
+ return partitions;
+ }
+
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public String error() {
+ return error;
+ }
+
+ @JsonProperty
+ public boolean success() {
+ return success;
+ }
+
+ }
+
+ @JsonPropertyOrder({ "timestamp", "name", "key", "value", "topic", "partition", "offset" })
+ public static class RecordData extends ShareConsumerEvent {
+
+ private final ConsumerRecord<String, String> record;
+
+ public RecordData(ConsumerRecord<String, String> record) {
+ this.record = record;
+ }
+
+ @Override
+ public String name() {
+ return "record_data";
+ }
+
+ @JsonProperty
+ public String topic() {
+ return record.topic();
+ }
+
+ @JsonProperty
+ public int partition() {
+ return record.partition();
+ }
+
+ @JsonProperty
Review Comment:
My mistake.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Infrastructure for system tests for the new share consumer client [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #18209:
URL: https://github.com/apache/kafka/pull/18209
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Infrastructure for system tests for the new share consumer client [kafka]
Posted by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org>.
chirag-wadhwa5 commented on code in PR #18209:
URL: https://github.com/apache/kafka/pull/18209#discussion_r1918822174
##########
tests/kafkatest/services/verifiable_share_consumer.py:
##########
@@ -0,0 +1,337 @@
+# 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.
+
+import json
+import os
+from ducktape.services.background_thread import BackgroundThreadService
+
+from kafkatest.directory_layout.kafka_path import KafkaPathResolverMixin
+from kafkatest.services.kafka import TopicPartition
+from kafkatest.services.verifiable_client import VerifiableClientMixin
+from kafkatest.version import DEV_BRANCH
+
+class ConsumerState:
+ Started = 1
+ Dead = 2
+
+class ShareConsumerEventHandler(object):
+
+ def __init__(self, node, idx, state=ConsumerState.Dead):
+ self.node = node
+ self.idx = idx
+ self.total_consumed = 0
+ self.total_acknowledged = 0
+ self.total_acknowledged_failed = 0
+ self.consumed_per_partition = {}
+ self.acknowledged_per_partition = {}
+ self.acknowledged_per_partition_failed = {}
+ self.state = state
+
+ def handle_shutdown_complete(self, node=None, logger=None):
+ self.state = ConsumerState.Dead
+ if node is not None and logger is not None:
+ logger.debug("Shut down %s" % node.account.hostname)
+
+ def handle_startup_complete(self, node, logger):
+ self.state = ConsumerState.Started
+ logger.debug("Started %s" % node.account.hostname)
+
+ def handle_offsets_acknowledged(self, event, node, logger):
+ if event["success"]:
+ self.total_acknowledged += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition[topic_partition] = self.acknowledged_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ else:
+ self.total_acknowledged_failed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.acknowledged_per_partition_failed[topic_partition] = self.acknowledged_per_partition_failed.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets acknowledged for %s" % (node.account.hostname))
+ logger.debug("Offset acknowledgement failed for: %s" % (node.account.hostname))
+
+ def handle_records_consumed(self, event, node, logger):
+ self.total_consumed += event["count"]
+ for share_partition_data in event["partitions"]:
+ topic_partition = TopicPartition(share_partition_data["topic"], share_partition_data["partition"])
+ self.consumed_per_partition[topic_partition] = self.consumed_per_partition.get(topic_partition, 0) + share_partition_data["count"]
+ logger.debug("Offsets consumed for %s" % (node.account.hostname))
+
+
+ def handle_kill_process(self, clean_shutdown):
+ # if the shutdown was clean, then we expect the explicit
+ # shutdown event from the consumer
+ if not clean_shutdown:
+ self.handle_shutdown_complete()
+
+class VerifiableShareConsumer(KafkaPathResolverMixin, VerifiableClientMixin, BackgroundThreadService):
+ """This service wraps org.apache.kafka.tools.VerifiableShareConsumer for use in
+ system testing.
+
+ NOTE: this class should be treated as a PUBLIC API. Downstream users use
+ this service both directly and through class extension, so care must be
+ taken to ensure compatibility.
+ """
+
+ PERSISTENT_ROOT = "/mnt/verifiable_share_consumer"
+ STDOUT_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stdout")
+ STDERR_CAPTURE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.stderr")
+ LOG_DIR = os.path.join(PERSISTENT_ROOT, "logs")
+ LOG_FILE = os.path.join(LOG_DIR, "verifiable_share_consumer.log")
+ LOG4J_CONFIG = os.path.join(PERSISTENT_ROOT, "tools-log4j.properties")
+ CONFIG_FILE = os.path.join(PERSISTENT_ROOT, "verifiable_share_consumer.properties")
+
+ logs = {
+ "verifiable_share_consumer_stdout": {
+ "path": STDOUT_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_stderr": {
+ "path": STDERR_CAPTURE,
+ "collect_default": False},
+ "verifiable_share_consumer_log": {
+ "path": LOG_FILE,
+ "collect_default": True}
+ }
+
+ def __init__(self, context, num_nodes, kafka, topic, group_id,
+ max_messages=-1, acknowledgement_mode="auto", offset_reset_strategy="",
+ version=DEV_BRANCH, stop_timeout_sec=60, log_level="INFO", jaas_override_variables=None,
+ on_record_consumed=None):
+ """
+ :param jaas_override_variables: A dict of variables to be used in the jaas.conf template file
+ """
+ super(VerifiableShareConsumer, self).__init__(context, num_nodes)
+ self.log_level = log_level
+ self.kafka = kafka
+ self.topic = topic
+ self.group_id = group_id
+ self.offset_reset_strategy = offset_reset_strategy
+ self.max_messages = max_messages
+ self.acknowledgement_mode = acknowledgement_mode
+ self.prop_file = ""
+ self.stop_timeout_sec = stop_timeout_sec
+ self.on_record_consumed = on_record_consumed
+
+ self.event_handlers = {}
+ self.jaas_override_variables = jaas_override_variables or {}
+
+ self.total_records_consumed = 0
+ self.total_records_acknowledged = 0
+ self.total_records_acknowledged_failed = 0
+ self.consumed_records_offsets = set()
+ # self.consumed_more_than_once = []
+
+ self.acknowledged_records_offsets = set()
+ # self.acknowledged_more_than_once = []
+ self.is_offset_reset_strategy_set = False
+
+ for node in self.nodes:
+ node.version = version
+
+ def java_class_name(self):
+ return "VerifiableShareConsumer"
+
+ def create_event_handler(self, idx, node):
+ return ShareConsumerEventHandler(node, idx)
+
+ def _worker(self, idx, node):
+ with self.lock:
+ self.event_handlers[node] = self.create_event_handler(idx, node)
+ handler = self.event_handlers[node]
+
+ node.account.ssh("mkdir -p %s" % VerifiableShareConsumer.PERSISTENT_ROOT, allow_fail=False)
+
+ # Create and upload log properties
+ log_config = self.render('tools_log4j.properties', log_file=VerifiableShareConsumer.LOG_FILE)
Review Comment:
Sure, I will do some investigations and get back to it. Thanks !
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org