mjsax commented on code in PR #19560:
URL: https://github.com/apache/kafka/pull/19560#discussion_r2060664760
##########
tests/kafkatest/tests/streams/base_streams_test.py:
##########
@@ -13,22 +13,40 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from ducktape.tests.test import Test
from ducktape.utils.util import wait_until
from kafkatest.services.verifiable_consumer import VerifiableConsumer
from kafkatest.services.verifiable_producer import VerifiableProducer
-from kafkatest.tests.kafka_test import KafkaTest
+from kafkatest.services.zookeeper import ZookeeperService
+from kafkatest.services.kafka import KafkaService, quorum
-class BaseStreamsTest(KafkaTest):
+class BaseStreamsTest(Test):
"""
Helper class that contains methods for producing and consuming
messages and verification of results from log files
Extends KafkaTest which manages setting up Kafka Cluster and Zookeeper
see tests/kafkatest/tests/kafka_test.py for more info
"""
- def __init__(self, test_context, topics, num_controllers=1,
num_brokers=3):
- super(BaseStreamsTest, self).__init__(test_context, num_controllers,
num_brokers, topics)
+ def __init__(self, test_context, topics, num_controllers=1, num_brokers=3):
+ self.num_zk = num_controllers
+ self.num_brokers = num_brokers
+ self.topics = topics
+
+ self.zk = ZookeeperService(test_context, self.num_zk) if
quorum.for_test(test_context) == quorum.zk else None
Review Comment:
Why this? I thought ZK is gone?
##########
tests/kafkatest/services/kafka/kafka.py:
##########
@@ -778,7 +781,14 @@ def prop_file(self, node):
if self.use_share_groups is not None and self.use_share_groups is True:
override_configs[config_property.SHARE_GROUP_ENABLE] =
str(self.use_share_groups)
-
override_configs[config_property.GROUP_COORDINATOR_REBALANCE_PROTOCOLS] =
'classic,consumer,share'
+ if self.use_streams_groups is True:
+ override_configs[config_property.UNSTABLE_API_VERSIONS_ENABLE]
= str(True)
+
override_configs[config_property.GROUP_COORDINATOR_REBALANCE_PROTOCOLS] =
'classic,consumer,share,streams'
+ else:
+
override_configs[config_property.GROUP_COORDINATOR_REBALANCE_PROTOCOLS] =
'classic,consumer,share'
+ elif self.use_streams_groups is True:
+ override_configs[config_property.UNSTABLE_API_VERSIONS_ENABLE] =
str(True)
+
override_configs[config_property.GROUP_COORDINATOR_REBALANCE_PROTOCOLS] =
'classic,consumer,streams'
Review Comment:
Would it be simpler to start with `enabledProtocols = 'classic,consumer'`
and do
```
if self.use_streams_groups is True:
enabledProtocols += ',streams`
```
and end with
```
override_configs[config_property.GROUP_COORDINATOR_REBALANCE_PROTOCOLS] =
enabledProtocols
```
Similar for share group? To avoid all this complicated nesting?
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
