You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jolshan (via GitHub)" <gi...@apache.org> on 2024/05/15 21:38:53 UTC
[PR] InvalidRequestException: ADD_PARTITIONS_TO_TXN with version 4 which is not enabled when upgrading from kafka [kafka]
jolshan opened a new pull request, #15971:
URL: https://github.com/apache/kafka/pull/15971
We weren't enabling discoverBrokerVersions to check the supported versions in the AddPartitionsToTxnManager. This means that any verification request (or any AddPartitionsToTxnRequest version) from a newer broker would fail when sending to an older broker.
The bulk of this change is adding additional transactions system tests for old versions.
One test upgrades the cluster completely. This didn't catch the issue but could be useful.
The other test forces a new broker to send a verification request to an older one. Without the discoverBrokerVersions change, all tests between mixed brokers failed. (We introduced a new request version in 3.8 -- which is a separate version from the one that caused the bug for 3.5 -> 3.6) With the addition, the tests all passed.
I also manually ran a test for 3.5 -> 3.6 since the issue there was slightly different and was caused by the unstableLatestVersion flag being enabled. The effect is the same and the fix also fixes this issue. 👍
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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] KAFKA-16992: InvalidRequestException: ADD_PARTITIONS_TO_TXN with version 4 which is not enabled when upgrading from kafka [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan merged PR #15971:
URL: https://github.com/apache/kafka/pull/15971
--
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] KAFKA-16992: InvalidRequestException: ADD_PARTITIONS_TO_TXN with version 4 which is not enabled when upgrading from kafka [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15971:
URL: https://github.com/apache/kafka/pull/15971#discussion_r1603400713
##########
core/src/main/java/kafka/server/NetworkUtils.java:
##########
@@ -82,7 +82,7 @@ public static NetworkClient buildNetworkClient(String prefix,
config.connectionSetupTimeoutMs(),
config.connectionSetupTimeoutMaxMs(),
time,
- false,
+ true,
Review Comment:
Yes, I believe this is the only usage now.
--
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] KAFKA-16992: InvalidRequestException: ADD_PARTITIONS_TO_TXN with version 4 which is not enabled when upgrading from kafka [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #15971:
URL: https://github.com/apache/kafka/pull/15971#discussion_r1603385828
##########
tests/kafkatest/tests/core/transactions_mixed_versions_test.py:
##########
@@ -0,0 +1,226 @@
+# 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 kafkatest.services.kafka import KafkaService, quorum, consumer_group
+from kafkatest.services.kafka.quorum import isolated_kraft
+from kafkatest.services.console_consumer import ConsoleConsumer
+from kafkatest.services.verifiable_producer import VerifiableProducer
+from kafkatest.services.transactional_message_copier import TransactionalMessageCopier
+from kafkatest.utils import is_int
+from kafkatest.utils.transactions_utils import create_and_start_copiers
+from kafkatest.version import LATEST_3_1, LATEST_3_2, LATEST_3_3, LATEST_3_4, LATEST_3_5, \
+ LATEST_3_6, LATEST_3_7, DEV_BRANCH, KafkaVersion, LATEST_STABLE_METADATA_VERSION
+
+from ducktape.tests.test import Test
+from ducktape.mark import matrix
+from ducktape.mark.resource import cluster
+from ducktape.utils.util import wait_until
+
+import time
+
+class TransactionsMixedVersionsTest(Test):
+ """Tests transactions by transactionally copying data from a source topic to
+ a destination between brokers with different image versions. This is how transactions work
+ while the cluster is undergoing an upgrade.
+ """
+ def __init__(self, test_context):
+ """:type test_context: ducktape.tests.test.TestContext"""
+ super(TransactionsMixedVersionsTest, self).__init__(test_context=test_context)
+
+ self.input_topic = "input-topic"
+ self.output_topic = "output-topic"
+
+ self.num_brokers = 3
+
+ self.replication_factor = 3
+
+ # Test parameters
+ self.num_input_partitions = 1
+ self.num_output_partitions = 1
+ self.num_seed_messages = 1000
+ self.transaction_size = 5
+
+ # The transaction timeout should be lower than the progress timeout, but at
+ # least as high as the request timeout (which is 30s by default).
+ self.transaction_timeout = 40000
+ self.progress_timeout_sec = 60
+ self.consumer_group = "transactions-test-consumer-group"
+
+ def seed_messages(self, topic, num_seed_messages):
+ seed_timeout_sec = 10000
+ seed_producer = VerifiableProducer(context=self.test_context,
+ num_nodes=1,
+ kafka=self.kafka,
+ topic=topic,
+ message_validator=is_int,
+ max_messages=num_seed_messages,
+ enable_idempotence=True)
+ seed_producer.start()
+ wait_until(lambda: seed_producer.num_acked >= num_seed_messages,
+ timeout_sec=seed_timeout_sec,
+ err_msg="Producer failed to produce messages %d in %ds." %\
+ (self.num_seed_messages, seed_timeout_sec))
+ return seed_producer.acked
+
+ def get_messages_from_topic(self, topic, num_messages, group_protocol):
+ consumer = self.start_consumer(topic, group_id="verifying_consumer", group_protocol=group_protocol)
+ return self.drain_consumer(consumer, num_messages)
+
+ def start_consumer(self, topic_to_read, group_id, group_protocol):
+ consumer = ConsoleConsumer(context=self.test_context,
+ num_nodes=1,
+ kafka=self.kafka,
+ topic=topic_to_read,
+ group_id=group_id,
+ message_validator=is_int,
+ from_beginning=True,
+ isolation_level="read_committed",
+ consumer_properties=consumer_group.maybe_set_group_protocol(group_protocol))
+ consumer.start()
+ # ensure that the consumer is up.
+ wait_until(lambda: (len(consumer.messages_consumed[1]) > 0) == True,
+ timeout_sec=60,
+ err_msg="Consumer failed to consume any messages for %ds" %\
+ 60)
+ return consumer
+
+ def drain_consumer(self, consumer, num_messages):
+ # wait until we read at least the expected number of messages.
+ # This is a safe check because both failure modes will be caught:
+ # 1. If we have 'num_seed_messages' but there are duplicates, then
+ # this is checked for later.
+ #
+ # 2. If we never reach 'num_seed_messages', then this will cause the
+ # test to fail.
+ wait_until(lambda: len(consumer.messages_consumed[1]) >= num_messages,
+ timeout_sec=90,
+ err_msg="Consumer consumed only %d out of %d messages in %ds" %\
+ (len(consumer.messages_consumed[1]), num_messages, 90))
+ consumer.stop()
+ return consumer.messages_consumed[1]
+
+ def copy_messages_transactionally(self, input_topic, output_topic,
+ num_copiers, num_messages_to_copy,
+ use_group_metadata, group_protocol):
+ """Copies messages transactionally from the seeded input topic to the
+ output topic.
+
+ This method also consumes messages in read_committed mode from the
+ output topic.
+
+ It returns the concurrently consumed messages.
+ """
+
+ copiers = create_and_start_copiers(test_context=self.test_context,
+ kafka=self.kafka,
+ consumer_group=self.consumer_group,
+ input_topic=input_topic,
+ output_topic=output_topic,
+ transaction_size=self.transaction_size,
+ transaction_timeout=self.transaction_timeout,
+ num_copiers=num_copiers,
+ use_group_metadata=use_group_metadata)
+ concurrent_consumer = self.start_consumer(output_topic,
+ group_id="concurrent_consumer",
+ group_protocol=group_protocol)
+
+ copier_timeout_sec = 120
+ for copier in copiers:
+ wait_until(lambda: copier.is_done,
+ timeout_sec=copier_timeout_sec,
+ err_msg="%s - Failed to copy all messages in %ds." %\
+ (copier.transactional_id, copier_timeout_sec))
+ self.logger.info("finished copying messages")
+
+ return self.drain_consumer(concurrent_consumer, num_messages_to_copy)
+
+ def setup_topics(self):
+ assignment = ":".join(map(str, [self.kafka.idx(node) for node in self.kafka.nodes]))
+ transaction_assignment = ",".join(map(str, [assignment[::-1]] * 50))
Review Comment:
Let me check if I got this right. You want the leaders of input/output to be on kafka-0 -- the broker with the latest version -- and the transaction state partition to be on the last broker. Is it correct?
##########
core/src/main/java/kafka/server/NetworkUtils.java:
##########
@@ -82,7 +82,7 @@ public static NetworkClient buildNetworkClient(String prefix,
config.connectionSetupTimeoutMs(),
config.connectionSetupTimeoutMaxMs(),
time,
- false,
+ true,
Review Comment:
This one is only used by the AddPartitionsToTxnManager, right?
##########
tests/kafkatest/utils/transactions_utils.py:
##########
@@ -0,0 +1,55 @@
+# Licensed 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 kafkatest.services.transactional_message_copier import TransactionalMessageCopier
+
+from ducktape.utils.util import wait_until
+
+def create_and_start_message_copier(test_context, kafka, consumer_group, input_topic, input_partition,
+ output_topic, transaction_size, transaction_timeout, transactional_id, use_group_metadata):
+ message_copier = TransactionalMessageCopier(
+ context=test_context,
+ num_nodes=1,
+ kafka=kafka,
+ transactional_id=transactional_id,
+ consumer_group=consumer_group,
+ input_topic=input_topic,
+ input_partition=input_partition,
+ output_topic=output_topic,
+ max_messages=-1,
+ transaction_size=transaction_size,
+ transaction_timeout=transaction_timeout,
+ use_group_metadata=use_group_metadata
+ )
+ message_copier.start()
+ wait_until(lambda: message_copier.alive(message_copier.nodes[0]),
+ timeout_sec=10,
+ err_msg="Message copier failed to start after 10 s")
+ return message_copier
+
+def create_and_start_copiers(test_context, kafka, consumer_group, input_topic, output_topic, transaction_size,
+ transaction_timeout, num_copiers, use_group_metadata):
+ copiers = []
+ for i in range(0, num_copiers):
+ copiers.append(create_and_start_message_copier(
+ test_context=test_context,
+ kafka=kafka,
+ consumer_group=consumer_group,
+ input_topic=input_topic,
+ output_topic=output_topic,
+ input_partition=i,
+ transaction_size=transaction_size,
+ transaction_timeout=transaction_timeout,
+ transactional_id="copier-" + str(i),
+ use_group_metadata=use_group_metadata
+ ))
+ return copiers
Review Comment:
nit: I would add an empty line at the end to be consistent.
--
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] KAFKA-16692: InvalidRequestException: ADD_PARTITIONS_TO_TXN with version 4 which is not enabled when upgrading from kafka [kafka]
Posted by "soarez (via GitHub)" <gi...@apache.org>.
soarez commented on PR #15971:
URL: https://github.com/apache/kafka/pull/15971#issuecomment-2121128256
> Not sure if we want to amend the commit name to fix it.
No, that could be very disruptive, not worth the trouble. It wasn't hard to find the JIRA despite the typo.
> I've picked to 3.6 and 3.7.
Nice. Thanks for closing the JIRA 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] KAFKA-16992: InvalidRequestException: ADD_PARTITIONS_TO_TXN with version 4 which is not enabled when upgrading from kafka [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #15971:
URL: https://github.com/apache/kafka/pull/15971#discussion_r1603399610
##########
tests/kafkatest/tests/core/transactions_mixed_versions_test.py:
##########
@@ -0,0 +1,226 @@
+# 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 kafkatest.services.kafka import KafkaService, quorum, consumer_group
+from kafkatest.services.kafka.quorum import isolated_kraft
+from kafkatest.services.console_consumer import ConsoleConsumer
+from kafkatest.services.verifiable_producer import VerifiableProducer
+from kafkatest.services.transactional_message_copier import TransactionalMessageCopier
+from kafkatest.utils import is_int
+from kafkatest.utils.transactions_utils import create_and_start_copiers
+from kafkatest.version import LATEST_3_1, LATEST_3_2, LATEST_3_3, LATEST_3_4, LATEST_3_5, \
+ LATEST_3_6, LATEST_3_7, DEV_BRANCH, KafkaVersion, LATEST_STABLE_METADATA_VERSION
+
+from ducktape.tests.test import Test
+from ducktape.mark import matrix
+from ducktape.mark.resource import cluster
+from ducktape.utils.util import wait_until
+
+import time
+
+class TransactionsMixedVersionsTest(Test):
+ """Tests transactions by transactionally copying data from a source topic to
+ a destination between brokers with different image versions. This is how transactions work
+ while the cluster is undergoing an upgrade.
+ """
+ def __init__(self, test_context):
+ """:type test_context: ducktape.tests.test.TestContext"""
+ super(TransactionsMixedVersionsTest, self).__init__(test_context=test_context)
+
+ self.input_topic = "input-topic"
+ self.output_topic = "output-topic"
+
+ self.num_brokers = 3
+
+ self.replication_factor = 3
+
+ # Test parameters
+ self.num_input_partitions = 1
+ self.num_output_partitions = 1
+ self.num_seed_messages = 1000
+ self.transaction_size = 5
+
+ # The transaction timeout should be lower than the progress timeout, but at
+ # least as high as the request timeout (which is 30s by default).
+ self.transaction_timeout = 40000
+ self.progress_timeout_sec = 60
+ self.consumer_group = "transactions-test-consumer-group"
+
+ def seed_messages(self, topic, num_seed_messages):
+ seed_timeout_sec = 10000
+ seed_producer = VerifiableProducer(context=self.test_context,
+ num_nodes=1,
+ kafka=self.kafka,
+ topic=topic,
+ message_validator=is_int,
+ max_messages=num_seed_messages,
+ enable_idempotence=True)
+ seed_producer.start()
+ wait_until(lambda: seed_producer.num_acked >= num_seed_messages,
+ timeout_sec=seed_timeout_sec,
+ err_msg="Producer failed to produce messages %d in %ds." %\
+ (self.num_seed_messages, seed_timeout_sec))
+ return seed_producer.acked
+
+ def get_messages_from_topic(self, topic, num_messages, group_protocol):
+ consumer = self.start_consumer(topic, group_id="verifying_consumer", group_protocol=group_protocol)
+ return self.drain_consumer(consumer, num_messages)
+
+ def start_consumer(self, topic_to_read, group_id, group_protocol):
+ consumer = ConsoleConsumer(context=self.test_context,
+ num_nodes=1,
+ kafka=self.kafka,
+ topic=topic_to_read,
+ group_id=group_id,
+ message_validator=is_int,
+ from_beginning=True,
+ isolation_level="read_committed",
+ consumer_properties=consumer_group.maybe_set_group_protocol(group_protocol))
+ consumer.start()
+ # ensure that the consumer is up.
+ wait_until(lambda: (len(consumer.messages_consumed[1]) > 0) == True,
+ timeout_sec=60,
+ err_msg="Consumer failed to consume any messages for %ds" %\
+ 60)
+ return consumer
+
+ def drain_consumer(self, consumer, num_messages):
+ # wait until we read at least the expected number of messages.
+ # This is a safe check because both failure modes will be caught:
+ # 1. If we have 'num_seed_messages' but there are duplicates, then
+ # this is checked for later.
+ #
+ # 2. If we never reach 'num_seed_messages', then this will cause the
+ # test to fail.
+ wait_until(lambda: len(consumer.messages_consumed[1]) >= num_messages,
+ timeout_sec=90,
+ err_msg="Consumer consumed only %d out of %d messages in %ds" %\
+ (len(consumer.messages_consumed[1]), num_messages, 90))
+ consumer.stop()
+ return consumer.messages_consumed[1]
+
+ def copy_messages_transactionally(self, input_topic, output_topic,
+ num_copiers, num_messages_to_copy,
+ use_group_metadata, group_protocol):
+ """Copies messages transactionally from the seeded input topic to the
+ output topic.
+
+ This method also consumes messages in read_committed mode from the
+ output topic.
+
+ It returns the concurrently consumed messages.
+ """
+
+ copiers = create_and_start_copiers(test_context=self.test_context,
+ kafka=self.kafka,
+ consumer_group=self.consumer_group,
+ input_topic=input_topic,
+ output_topic=output_topic,
+ transaction_size=self.transaction_size,
+ transaction_timeout=self.transaction_timeout,
+ num_copiers=num_copiers,
+ use_group_metadata=use_group_metadata)
+ concurrent_consumer = self.start_consumer(output_topic,
+ group_id="concurrent_consumer",
+ group_protocol=group_protocol)
+
+ copier_timeout_sec = 120
+ for copier in copiers:
+ wait_until(lambda: copier.is_done,
+ timeout_sec=copier_timeout_sec,
+ err_msg="%s - Failed to copy all messages in %ds." %\
+ (copier.transactional_id, copier_timeout_sec))
+ self.logger.info("finished copying messages")
+
+ return self.drain_consumer(concurrent_consumer, num_messages_to_copy)
+
+ def setup_topics(self):
+ assignment = ":".join(map(str, [self.kafka.idx(node) for node in self.kafka.nodes]))
+ transaction_assignment = ",".join(map(str, [assignment[::-1]] * 50))
Review Comment:
Yes we want to send a verification request to a broker that might not support it.
##########
tests/kafkatest/tests/core/transactions_mixed_versions_test.py:
##########
@@ -0,0 +1,226 @@
+# 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 kafkatest.services.kafka import KafkaService, quorum, consumer_group
+from kafkatest.services.kafka.quorum import isolated_kraft
+from kafkatest.services.console_consumer import ConsoleConsumer
+from kafkatest.services.verifiable_producer import VerifiableProducer
+from kafkatest.services.transactional_message_copier import TransactionalMessageCopier
+from kafkatest.utils import is_int
+from kafkatest.utils.transactions_utils import create_and_start_copiers
+from kafkatest.version import LATEST_3_1, LATEST_3_2, LATEST_3_3, LATEST_3_4, LATEST_3_5, \
+ LATEST_3_6, LATEST_3_7, DEV_BRANCH, KafkaVersion, LATEST_STABLE_METADATA_VERSION
+
+from ducktape.tests.test import Test
+from ducktape.mark import matrix
+from ducktape.mark.resource import cluster
+from ducktape.utils.util import wait_until
+
+import time
+
+class TransactionsMixedVersionsTest(Test):
+ """Tests transactions by transactionally copying data from a source topic to
+ a destination between brokers with different image versions. This is how transactions work
+ while the cluster is undergoing an upgrade.
+ """
+ def __init__(self, test_context):
+ """:type test_context: ducktape.tests.test.TestContext"""
+ super(TransactionsMixedVersionsTest, self).__init__(test_context=test_context)
+
+ self.input_topic = "input-topic"
+ self.output_topic = "output-topic"
+
+ self.num_brokers = 3
+
+ self.replication_factor = 3
+
+ # Test parameters
+ self.num_input_partitions = 1
+ self.num_output_partitions = 1
+ self.num_seed_messages = 1000
+ self.transaction_size = 5
+
+ # The transaction timeout should be lower than the progress timeout, but at
+ # least as high as the request timeout (which is 30s by default).
+ self.transaction_timeout = 40000
+ self.progress_timeout_sec = 60
+ self.consumer_group = "transactions-test-consumer-group"
+
+ def seed_messages(self, topic, num_seed_messages):
+ seed_timeout_sec = 10000
+ seed_producer = VerifiableProducer(context=self.test_context,
+ num_nodes=1,
+ kafka=self.kafka,
+ topic=topic,
+ message_validator=is_int,
+ max_messages=num_seed_messages,
+ enable_idempotence=True)
+ seed_producer.start()
+ wait_until(lambda: seed_producer.num_acked >= num_seed_messages,
+ timeout_sec=seed_timeout_sec,
+ err_msg="Producer failed to produce messages %d in %ds." %\
+ (self.num_seed_messages, seed_timeout_sec))
+ return seed_producer.acked
+
+ def get_messages_from_topic(self, topic, num_messages, group_protocol):
+ consumer = self.start_consumer(topic, group_id="verifying_consumer", group_protocol=group_protocol)
+ return self.drain_consumer(consumer, num_messages)
+
+ def start_consumer(self, topic_to_read, group_id, group_protocol):
+ consumer = ConsoleConsumer(context=self.test_context,
+ num_nodes=1,
+ kafka=self.kafka,
+ topic=topic_to_read,
+ group_id=group_id,
+ message_validator=is_int,
+ from_beginning=True,
+ isolation_level="read_committed",
+ consumer_properties=consumer_group.maybe_set_group_protocol(group_protocol))
+ consumer.start()
+ # ensure that the consumer is up.
+ wait_until(lambda: (len(consumer.messages_consumed[1]) > 0) == True,
+ timeout_sec=60,
+ err_msg="Consumer failed to consume any messages for %ds" %\
+ 60)
+ return consumer
+
+ def drain_consumer(self, consumer, num_messages):
+ # wait until we read at least the expected number of messages.
+ # This is a safe check because both failure modes will be caught:
+ # 1. If we have 'num_seed_messages' but there are duplicates, then
+ # this is checked for later.
+ #
+ # 2. If we never reach 'num_seed_messages', then this will cause the
+ # test to fail.
+ wait_until(lambda: len(consumer.messages_consumed[1]) >= num_messages,
+ timeout_sec=90,
+ err_msg="Consumer consumed only %d out of %d messages in %ds" %\
+ (len(consumer.messages_consumed[1]), num_messages, 90))
+ consumer.stop()
+ return consumer.messages_consumed[1]
+
+ def copy_messages_transactionally(self, input_topic, output_topic,
+ num_copiers, num_messages_to_copy,
+ use_group_metadata, group_protocol):
+ """Copies messages transactionally from the seeded input topic to the
+ output topic.
+
+ This method also consumes messages in read_committed mode from the
+ output topic.
+
+ It returns the concurrently consumed messages.
+ """
+
+ copiers = create_and_start_copiers(test_context=self.test_context,
+ kafka=self.kafka,
+ consumer_group=self.consumer_group,
+ input_topic=input_topic,
+ output_topic=output_topic,
+ transaction_size=self.transaction_size,
+ transaction_timeout=self.transaction_timeout,
+ num_copiers=num_copiers,
+ use_group_metadata=use_group_metadata)
+ concurrent_consumer = self.start_consumer(output_topic,
+ group_id="concurrent_consumer",
+ group_protocol=group_protocol)
+
+ copier_timeout_sec = 120
+ for copier in copiers:
+ wait_until(lambda: copier.is_done,
+ timeout_sec=copier_timeout_sec,
+ err_msg="%s - Failed to copy all messages in %ds." %\
+ (copier.transactional_id, copier_timeout_sec))
+ self.logger.info("finished copying messages")
+
+ return self.drain_consumer(concurrent_consumer, num_messages_to_copy)
+
+ def setup_topics(self):
+ assignment = ":".join(map(str, [self.kafka.idx(node) for node in self.kafka.nodes]))
+ transaction_assignment = ",".join(map(str, [assignment[::-1]] * 50))
Review Comment:
Yes we want to attempt to send a verification request to a broker that might not support it.
--
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] KAFKA-16692: InvalidRequestException: ADD_PARTITIONS_TO_TXN with version 4 which is not enabled when upgrading from kafka [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #15971:
URL: https://github.com/apache/kafka/pull/15971#issuecomment-2120961394
Yes. I just noticed this too 🤦♀️ Not sure if we want to amend the commit name to fix it.
I've picked to 3.6 and 3.7.
--
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] KAFKA-16992: InvalidRequestException: ADD_PARTITIONS_TO_TXN with version 4 which is not enabled when upgrading from kafka [kafka]
Posted by "soarez (via GitHub)" <gi...@apache.org>.
soarez commented on PR #15971:
URL: https://github.com/apache/kafka/pull/15971#issuecomment-2120674801
I think there was typo on the PR and commit title, it should refer [KAFKA-16692](https://issues.apache.org/jira/browse/KAFKA-16692).
--
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