First draft of system tests for the new share consumer client #18209
Conversation
| @@ -288,6 +288,8 @@ | |||
| files="VerifiableConsumer.java"/> | |||
| <suppress id="dontUseSystemExit" | |||
| files="VerifiableProducer.java"/> | |||
| <suppress id="dontUseSystemExit" | |||
| files="VerifiableShareGroup.java"/> | |||
There was a problem hiding this comment.
VerifiableShareGroup doesn't exist. Do you mean VerifiableShareConsumer?
| @@ -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 | |||
There was a problem hiding this comment.
These references to VerifiableShareGroup and ShareGroup should probably be VerifiableShareConsumer and ShareConsumer.
AndrewJSchofield
added
KIP-932
|
|
||
| private static ArgumentParser argParser() { | ||
| ArgumentParser parser = ArgumentParsers | ||
| .newArgumentParser("verifiable-share-group") |
There was a problem hiding this comment.
"verifiable-share-consumer"
| .setDefault("") | ||
| .type(String.class) | ||
| .dest("offsetResetStrategy") | ||
| .help("Set share group reset strategy (must be either 'earliest' or 'latest'"); |
There was a problem hiding this comment.
nit: Missing ')' int he string.
| } | ||
| } | ||
| } catch (WakeupException e) { | ||
| out.println("caught wakeup exception: " + e); |
There was a problem hiding this comment.
Remove this line please.
| // 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); |
| printJson(new RecordData(record)); | ||
| } | ||
| } | ||
|
|
| // 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()); |
There was a problem hiding this comment.
If we want to write out these pieces of information, I suggest a better way is appropriate. I'd remove these two printlns.
| private final boolean success; | ||
|
|
||
| public OffsetsAcknowledged(long count, List<AcknowledgedData> partitions, String error, boolean success) { | ||
| this.count=count; |
| @@ -0,0 +1,35 @@ | |||
| #!/bin/bash | |||
There was a problem hiding this 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.
| @@ -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 | |||
There was a problem hiding this 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.
| 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 |
There was a problem hiding this comment.
I think this should be applicable for DEV_BRANCH and V_4_1_0 only.
This is the first draft for System Tests for the new share consumer client introduced as part of KIP-932