Skip to content

KAFKA-15913: Migrate async consumer tests to mocks #14930

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
lucasbru merged 24 commits into from
Dec 18, 2023
Merged

KAFKA-15913: Migrate async consumer tests to mocks #14930

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
8159e51
migrate epoch update to mocks
lucasbru Dec 5, 2023
5b95d03
migrate fenced exception
lucasbru Dec 5, 2023
7dcd8ae
migrate committed tests
lucasbru Dec 5, 2023
966c18f
migrate subscribed tests
lucasbru Dec 5, 2023
407a85b
migrate offset tests
lucasbru Dec 5, 2023
b40e5c9
migrate wakeup tests
lucasbru Dec 5, 2023
4ff9b57
migrate assing tests
lucasbru Dec 5, 2023
283cf51
Add test testCommitAsync_NullCallback as testCommitAsyncWithNullCallback
cadonna Dec 6, 2023
53b0fdf
use most of the normal construction code for testing and migrate conf…
lucasbru Dec 7, 2023
5630c42
fix rebase errors, port remote assignor and group metadata tests
lucasbru Dec 8, 2023
ab35f4e
port long poll test
lucasbru Dec 11, 2023
4d3fea9
migrate commit callback tests
lucasbru Dec 11, 2023
b00135e
migrate wakeup committed
lucasbru Dec 11, 2023
90a1f5d
reorder
lucasbru Dec 11, 2023
60037dd
updates
lucasbru Dec 11, 2023
6377a9d
checkstyle
lucasbru Dec 11, 2023
66a0c15
PR comments
lucasbru Dec 11, 2023
c6a5ee0
Merge remote-tracking branch 'upstream/trunk' into ctr_migrate_test
lucasbru Dec 11, 2023
e6430ca
remove unused code.
lucasbru Dec 12, 2023
9648ca3
Merge remote-tracking branch 'upstream/trunk' into ctr_migrate_test
lucasbru Dec 12, 2023
59e385b
optimize imports
lucasbru Dec 12, 2023
dde99b3
checkstyle
lucasbru Dec 12, 2023
dc4f96d
Merge remote-tracking branch 'upstream/trunk' into ctr_migrate_test
lucasbru Dec 18, 2023
5e734fe
merge fixes
lucasbru Dec 18, 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
118 changes: 67 additions & 51 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,6 @@
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.Set;
Expand Down Expand Up @@ -287,12 +286,26 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
AsyncKafkaConsumer(final ConsumerConfig config,
final Deserializer<K> keyDeserializer,
final Deserializer<V> valueDeserializer) {
this(config, keyDeserializer, valueDeserializer, new LinkedBlockingQueue<>());
this(
config,
keyDeserializer,
valueDeserializer,
Time.SYSTEM,
ApplicationEventHandler::new,
FetchCollector::new,
ConsumerMetadata::new,
new LinkedBlockingQueue<>()
);
}

// Visible for testing
AsyncKafkaConsumer(final ConsumerConfig config,
final Deserializer<K> keyDeserializer,
final Deserializer<V> valueDeserializer,
final Time time,
final ApplicationEventHandlerFactory applicationEventHandlerFactory,
final FetchCollectorFactory<K, V> fetchCollectorFactory,
final ConsumerMetadataFactory metadataFactory,
final LinkedBlockingQueue<BackgroundEvent> backgroundEventQueue) {
try {
GroupRebalanceConfig groupRebalanceConfig = new GroupRebalanceConfig(
Expand All @@ -305,7 +318,7 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {

log.debug("Initializing the Kafka consumer");
this.defaultApiTimeoutMs = config.getInt(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG);
this.time = Time.SYSTEM;
this.time = time;
List<MetricsReporter> reporters = CommonClientConfigs.metricsReporters(clientId, config);
this.clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, config);
this.clientTelemetryReporter.ifPresent(reporters::add);
Expand All @@ -319,7 +332,7 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
ClusterResourceListeners clusterResourceListeners = ClientUtils.configureClusterResourceListeners(metrics.reporters(),
interceptorList,
Arrays.asList(deserializers.keyDeserializer, deserializers.valueDeserializer));
this.metadata = new ConsumerMetadata(config, subscriptions, logContext, clusterResourceListeners);
this.metadata = metadataFactory.build(config, subscriptions, logContext, clusterResourceListeners);
final List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config);
metadata.bootstrap(addresses);

Expand Down Expand Up @@ -360,7 +373,8 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
metadata,
applicationEventQueue,
requestManagersSupplier);
this.applicationEventHandler = new ApplicationEventHandler(logContext,
this.applicationEventHandler = applicationEventHandlerFactory.build(
logContext,
time,
applicationEventQueue,
applicationEventProcessorSupplier,
Expand Down Expand Up @@ -391,7 +405,7 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
this.groupMetadata = initializeGroupMetadata(config, groupRebalanceConfig);

// The FetchCollector is only used on the application thread.
this.fetchCollector = new FetchCollector<>(logContext,
this.fetchCollector = fetchCollectorFactory.build(logContext,
metadata,
subscriptions,
fetchConfig,
Expand Down Expand Up @@ -420,49 +434,6 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
}

// Visible for testing
AsyncKafkaConsumer(LogContext logContext,
String clientId,
Deserializers<K, V> deserializers,
FetchBuffer fetchBuffer,
FetchCollector<K, V> fetchCollector,
ConsumerInterceptors<K, V> interceptors,
Time time,
ApplicationEventHandler applicationEventHandler,
BlockingQueue<BackgroundEvent> backgroundEventQueue,
ConsumerRebalanceListenerInvoker rebalanceListenerInvoker,
Metrics metrics,
SubscriptionState subscriptions,
ConsumerMetadata metadata,
long retryBackoffMs,
int defaultApiTimeoutMs,
List<ConsumerPartitionAssignor> assignors,
String groupId) {
this.log = logContext.logger(getClass());
this.subscriptions = subscriptions;
this.clientId = clientId;
this.fetchBuffer = fetchBuffer;
this.fetchCollector = fetchCollector;
this.isolationLevel = IsolationLevel.READ_UNCOMMITTED;
this.interceptors = Objects.requireNonNull(interceptors);
this.time = time;
this.backgroundEventProcessor = new BackgroundEventProcessor(
logContext,
backgroundEventQueue,
applicationEventHandler,
rebalanceListenerInvoker
);
this.metrics = metrics;
this.groupMetadata = initializeGroupMetadata(groupId, Optional.empty());
this.metadata = metadata;
this.retryBackoffMs = retryBackoffMs;
this.defaultApiTimeoutMs = defaultApiTimeoutMs;
this.deserializers = deserializers;
this.applicationEventHandler = applicationEventHandler;
this.assignors = assignors;
this.kafkaConsumerMetrics = new KafkaConsumerMetrics(metrics, "consumer");
this.clientTelemetryReporter = Optional.empty();
}

AsyncKafkaConsumer(LogContext logContext,
Time time,
ConsumerConfig config,
Expand Down Expand Up @@ -563,6 +534,47 @@ private void process(final ConsumerRebalanceListenerCallbackNeededEvent event) {
);
}

// auxiliary interface for testing
interface ApplicationEventHandlerFactory {

ApplicationEventHandler build(
final LogContext logContext,
final Time time,
final BlockingQueue<ApplicationEvent> applicationEventQueue,
final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier,
final Supplier<NetworkClientDelegate> networkClientDelegateSupplier,
final Supplier<RequestManagers> requestManagersSupplier
);

}

// auxiliary interface for testing
interface FetchCollectorFactory<K, V> {

FetchCollector<K, V> build(
final LogContext logContext,
final ConsumerMetadata metadata,
final SubscriptionState subscriptions,
final FetchConfig fetchConfig,
final Deserializers<K, V> deserializers,
final FetchMetricsManager metricsManager,
final Time time
);

}

// auxiliary interface for testing
interface ConsumerMetadataFactory {

ConsumerMetadata build(
final ConsumerConfig config,
final SubscriptionState subscriptions,
final LogContext logContext,
final ClusterResourceListeners clusterResourceListeners
);

}

private Optional<ConsumerGroupMetadata> initializeGroupMetadata(final ConsumerConfig config,
final GroupRebalanceConfig groupRebalanceConfig) {
final Optional<ConsumerGroupMetadata> groupMetadata = initializeGroupMetadata(
Expand Down Expand Up @@ -1756,8 +1768,7 @@ private void maybeThrowFencedInstanceException() {
}
}

// Visible for testing
void maybeInvokeCommitCallbacks() {
private void maybeInvokeCommitCallbacks() {
if (callbacks() > 0) {
invoker.executeCallbacks();
}
Expand All @@ -1768,6 +1779,11 @@ int callbacks() {
return invoker.callbackQueue.size();
}

// Visible for testing
SubscriptionState subscriptions() {
return subscriptions;
}
Comment on lines +1782 to +1785
Copy link
Copy Markdown
Member

@cadonna cadonna Dec 11, 2023

Choose a reason for hiding this comment

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

Is there a way to avoid this?

Copy link
Copy Markdown
Member Author

@lucasbru lucasbru Dec 11, 2023

Choose a reason for hiding this comment

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

  • I can add callback to the constructor that registers the SubscriptionState with the calling context (the test)
  • I can add another factory to inject our own instance of SubscriptionState


/**
* Utility class that helps the application thread to invoke user registered {@link OffsetCommitCallback}. This is
* achieved by having the background thread register a {@link OffsetCommitCallbackTask} to the invoker upon the
Expand Down
Loading