Skip to content

KAFKA-14905: Reduce flakiness in MM2 ForwardingAdmin test due to admin timeouts #13575

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jlprat merged 2 commits into from
Apr 21, 2023
Merged

KAFKA-14905: Reduce flakiness in MM2 ForwardingAdmin test due to admin timeouts #13575

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c7becca
KAFKA-14905: Reduce flakiness in MM2 ForwardingAdmin test due to admi…
gharris1727 Apr 18, 2023
38329c3
fixup: use whenComplete to avoid infinite blocking
gharris1727 Apr 21, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 24 additions & 38 deletions ...a/org/apache/kafka/connect/mirror/clients/admin/FakeForwardingAdminWithLocalMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,17 +37,13 @@

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/** Customised ForwardingAdmin for testing only.
* The class create/alter topics, partitions and ACLs in Kafka then store metadata in {@link FakeLocalMetadataStore}.
* */

public class FakeForwardingAdminWithLocalMetadata extends ForwardingAdmin {
private static final Logger log = LoggerFactory.getLogger(FakeForwardingAdminWithLocalMetadata.class);
private final long timeout = 1000L;

@jlprat jlprat Apr 21, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Any reason why removing the timeout completely instead of increasing it to some abundantly high number of seconds?
My time at Akka tells me: never wait forever; always set a timeout, even if it's high.
What do you think about putting something like 15 seconds? This way if something goes wrong and the future never finished we won't be hanging there forever.

@gharris1727 gharris1727 Apr 21, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I didn't just raise the timeout because the MirrorSourceTask that makes these calls does not have a timeout that I could predictably meet. The task appears to either use get() or whenComplete on the futures returned by these calls.

I think your critique is correct though, and I've rewritten this to use whenComplete() to get the results of the futures, instead of get().

@jlprat jlprat Apr 21, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I like the whenComplete constructs, I think they are more elegant as well.


public FakeForwardingAdminWithLocalMetadata(Map<String, Object> configs) {
super(configs);
Expand All @@ -56,69 +52,59 @@ public FakeForwardingAdminWithLocalMetadata(Map<String, Object> configs) {
@Override
public CreateTopicsResult createTopics(Collection<NewTopic> newTopics, CreateTopicsOptions options) {
CreateTopicsResult createTopicsResult = super.createTopics(newTopics, options);
newTopics.forEach(newTopic -> {
try {
log.info("Add topic '{}' to cluster and metadata store", newTopic);
// Wait for topic to be created before edit the fake local store
createTopicsResult.values().get(newTopic.name()).get(timeout, TimeUnit.MILLISECONDS);
newTopics.forEach(newTopic -> createTopicsResult.values().get(newTopic.name()).whenComplete((ignored, error) -> {
if (error == null) {
FakeLocalMetadataStore.addTopicToLocalMetadataStore(newTopic);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
if (e.getCause() instanceof TopicExistsException) {
log.warn("Topic '{}' already exists. Update the local metadata store if absent", newTopic.name());
FakeLocalMetadataStore.addTopicToLocalMetadataStore(newTopic);
} else
log.error(e.getMessage());
} else if (error.getCause() instanceof TopicExistsException) {
log.warn("Topic '{}' already exists. Update the local metadata store if absent", newTopic.name());
FakeLocalMetadataStore.addTopicToLocalMetadataStore(newTopic);
} else {
log.error("Unable to intercept admin client operation", error);
}
});
}));
return createTopicsResult;
}

@Override
public CreatePartitionsResult createPartitions(Map<String, NewPartitions> newPartitions, CreatePartitionsOptions options) {
CreatePartitionsResult createPartitionsResult = super.createPartitions(newPartitions, options);
newPartitions.forEach((topic, newPartition) -> {
try {
// Wait for topic partition to be created before edit the fake local store
createPartitionsResult.values().get(topic).get(timeout, TimeUnit.MILLISECONDS);
newPartitions.forEach((topic, newPartition) -> createPartitionsResult.values().get(topic).whenComplete((ignored, error) -> {
if (error == null) {
FakeLocalMetadataStore.updatePartitionCount(topic, newPartition.totalCount());
} catch (InterruptedException | ExecutionException | TimeoutException e) {
log.error(e.getMessage());
} else {
log.error("Unable to intercept admin client operation", error);
}
});
}));
return createPartitionsResult;
}

@Deprecated
@Override
public AlterConfigsResult alterConfigs(Map<ConfigResource, Config> configs, AlterConfigsOptions options) {
AlterConfigsResult alterConfigsResult = super.alterConfigs(configs, options);
configs.forEach((configResource, newConfigs) -> {
try {
configs.forEach((configResource, newConfigs) -> alterConfigsResult.values().get(configResource).whenComplete((ignored, error) -> {
if (error == null) {
if (configResource.type() == ConfigResource.Type.TOPIC) {
// Wait for config to be altered before edit the fake local store
alterConfigsResult.values().get(configResource).get(timeout, TimeUnit.MILLISECONDS);
FakeLocalMetadataStore.updateTopicConfig(configResource.name(), newConfigs);
}
} catch (InterruptedException | ExecutionException | TimeoutException e) {
log.error(e.getMessage());
} else {
log.error("Unable to intercept admin client operation", error);
}
});
}));
return alterConfigsResult;
}


@Override
public CreateAclsResult createAcls(Collection<AclBinding> acls, CreateAclsOptions options) {
CreateAclsResult aclsResult = super.createAcls(acls, options);
try {
// Wait for acls to be created before edit the fake local store
aclsResult.all().get(timeout, TimeUnit.MILLISECONDS);
acls.forEach(aclBinding -> {
aclsResult.values().forEach((aclBinding, future) -> future.whenComplete((ignored, error) -> {
if (error == null) {
FakeLocalMetadataStore.addACLs(aclBinding.entry().principal(), aclBinding);
});
} catch (InterruptedException | ExecutionException | TimeoutException e) {
log.error(e.getMessage());
}
} else {
log.error("Unable to intercept admin client operation", error);
}
}));
return aclsResult;
}
}