diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java new file mode 100644 index 0000000000000..546e0e8e76fb1 --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java @@ -0,0 +1,230 @@ +/* + * 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. + */ +package org.apache.kafka.clients.consumer.internals; + +import org.apache.kafka.clients.CommonClientConfigs; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.internals.events.ApplicationEvent; +import org.apache.kafka.clients.consumer.internals.events.BackgroundEvent; +import org.apache.kafka.clients.consumer.internals.events.NoopApplicationEvent; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.errors.WakeupException; +import org.apache.kafka.common.metrics.Metrics; +import org.apache.kafka.common.utils.KafkaThread; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; +import org.slf4j.Logger; + +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Background thread runnable that consumes {@code ApplicationEvent} and + * produces {@code BackgroundEvent}. It uses an event loop to consume and + * produce events, and poll the network client to handle network IO. + *

+ * It holds a reference to the {@link SubscriptionState}, which is + * initialized by the polling thread. + */ +public class DefaultBackgroundThread extends KafkaThread { + private static final String BACKGROUND_THREAD_NAME = + "consumer_background_thread"; + private final Time time; + private final Logger log; + private final BlockingQueue applicationEventQueue; + private final BlockingQueue backgroundEventQueue; + private final ConsumerNetworkClient networkClient; + private final SubscriptionState subscriptions; + private final ConsumerMetadata metadata; + private final Metrics metrics; + private final ConsumerConfig config; + + private String clientId; + private long retryBackoffMs; + private int heartbeatIntervalMs; + private boolean running; + private Optional inflightEvent = Optional.empty(); + private final AtomicReference> exception = + new AtomicReference<>(Optional.empty()); + + public DefaultBackgroundThread(final ConsumerConfig config, + final LogContext logContext, + final BlockingQueue applicationEventQueue, + final BlockingQueue backgroundEventQueue, + final SubscriptionState subscriptions, + final ConsumerMetadata metadata, + final ConsumerNetworkClient networkClient, + final Metrics metrics) { + this( + Time.SYSTEM, + config, + logContext, + applicationEventQueue, + backgroundEventQueue, + subscriptions, + metadata, + networkClient, + metrics + ); + } + + public DefaultBackgroundThread(final Time time, + final ConsumerConfig config, + final LogContext logContext, + final BlockingQueue applicationEventQueue, + final BlockingQueue backgroundEventQueue, + final SubscriptionState subscriptions, + final ConsumerMetadata metadata, + final ConsumerNetworkClient networkClient, + final Metrics metrics) { + super(BACKGROUND_THREAD_NAME, true); + try { + this.time = time; + this.log = logContext.logger(DefaultBackgroundThread.class); + this.applicationEventQueue = applicationEventQueue; + this.backgroundEventQueue = backgroundEventQueue; + this.config = config; + setConfig(); + this.inflightEvent = Optional.empty(); + // subscriptionState is initialized by the polling thread + this.subscriptions = subscriptions; + this.metadata = metadata; + this.networkClient = networkClient; + this.metrics = metrics; + this.running = true; + } catch (final Exception e) { + // now propagate the exception + close(); + throw new KafkaException("Failed to construct background processor", e); + } + } + + private void setConfig() { + this.retryBackoffMs = this.config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG); + this.clientId = config.getString(CommonClientConfigs.CLIENT_ID_CONFIG); + this.heartbeatIntervalMs = config.getInt(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG); + } + + @Override + public void run() { + try { + log.debug("Background thread started"); + while (running) { + try { + runOnce(); + } catch (final WakeupException e) { + log.debug( + "Exception thrown, background thread won't terminate", + e + ); + // swallow the wakeup exception to prevent killing the + // background thread. + } + } + } catch (final Throwable t) { + log.error( + "The background thread failed due to unexpected error", + t + ); + if (t instanceof RuntimeException) + this.exception.set(Optional.of((RuntimeException) t)); + else + this.exception.set(Optional.of(new RuntimeException(t))); + } finally { + close(); + log.debug("{} closed", getClass()); + } + } + + /** + * Process event from a single poll + */ + void runOnce() { + this.inflightEvent = maybePollEvent(); + if (this.inflightEvent.isPresent()) { + log.debug("processing application event: {}", this.inflightEvent); + } + if (this.inflightEvent.isPresent() && maybeConsumeInflightEvent(this.inflightEvent.get())) { + // clear inflight event upon successful consumption + this.inflightEvent = Optional.empty(); + } + + // if there are pending events to process, poll then continue without + // blocking. + if (!applicationEventQueue.isEmpty() || inflightEvent.isPresent()) { + networkClient.poll(time.timer(0)); + return; + } + // if there are no events to process, poll until timeout. The timeout + // will be the minimum of the requestTimeoutMs, nextHeartBeatMs, and + // nextMetadataUpdate. See NetworkClient.poll impl. + networkClient.poll(time.timer(timeToNextHeartbeatMs(time.milliseconds()))); + } + + private long timeToNextHeartbeatMs(final long nowMs) { + // TODO: implemented when heartbeat is added to the impl + return 100; + } + + private Optional maybePollEvent() { + if (this.inflightEvent.isPresent() || this.applicationEventQueue.isEmpty()) { + return this.inflightEvent; + } + return Optional.ofNullable(this.applicationEventQueue.poll()); + } + + /** + * ApplicationEvent are consumed here. + * + * @param event an {@link ApplicationEvent} + * @return true when successfully consumed the event. + */ + private boolean maybeConsumeInflightEvent(final ApplicationEvent event) { + log.debug("try consuming event: {}", Optional.ofNullable(event)); + Objects.requireNonNull(event); + return event.process(); + } + + /** + * Processes {@link NoopApplicationEvent} and equeue a + * {@link NoopBackgroundEvent}. This is intentionally left here for + * demonstration purpose. + * + * @param event a {@link NoopApplicationEvent} + */ + private void process(final NoopApplicationEvent event) { + backgroundEventQueue.add(new NoopBackgroundEvent(event.message)); + } + + public boolean isRunning() { + return this.running; + } + + public void wakeup() { + networkClient.wakeup(); + } + + public void close() { + this.running = false; + this.wakeup(); + Utils.closeQuietly(networkClient, "consumer network client"); + Utils.closeQuietly(metadata, "consumer metadata client"); + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandler.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandler.java index ec1368ed801f8..3a5a43abf7d73 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandler.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandler.java @@ -16,45 +16,183 @@ */ package org.apache.kafka.clients.consumer.internals; +import org.apache.kafka.clients.ApiVersions; +import org.apache.kafka.clients.ClientUtils; +import org.apache.kafka.clients.NetworkClient; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.internals.events.ApplicationEvent; import org.apache.kafka.clients.consumer.internals.events.BackgroundEvent; import org.apache.kafka.clients.consumer.internals.events.EventHandler; +import org.apache.kafka.common.internals.ClusterResourceListeners; +import org.apache.kafka.common.metrics.Metrics; +import org.apache.kafka.common.metrics.Sensor; +import org.apache.kafka.common.network.ChannelBuilder; +import org.apache.kafka.common.network.Selector; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Time; +import java.net.InetSocketAddress; +import java.util.List; import java.util.Optional; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; /** - * This class interfaces the KafkaConsumer and the background thread. It allows the caller to enqueue {@link ApplicationEvent} - * to be consumed by the background thread and poll {@linkBackgroundEvent} produced by the background thread. + * An {@code EventHandler} that uses a single background thread to consume {@code ApplicationEvent} and produce + * {@code BackgroundEvent} from the {@ConsumerBackgroundThread}. */ public class DefaultEventHandler implements EventHandler { - private final BlockingQueue applicationEvents; - private final BlockingQueue backgroundEvents; + private static final String METRIC_GRP_PREFIX = "consumer"; + private final BlockingQueue applicationEventQueue; + private final BlockingQueue backgroundEventQueue; + private final DefaultBackgroundThread backgroundThread; - public DefaultEventHandler() { - this.applicationEvents = new LinkedBlockingQueue<>(); - this.backgroundEvents = new LinkedBlockingQueue<>(); - // TODO: a concreted implementation of how requests are being consumed, and how responses are being produced. + + public DefaultEventHandler(final Time time, + final ConsumerConfig config, + final LogContext logContext, + final BlockingQueue applicationEventQueue, + final BlockingQueue backgroundEventQueue, + final SubscriptionState subscriptionState, + final ApiVersions apiVersions, + final Metrics metrics, + final ClusterResourceListeners clusterResourceListeners, + final Sensor fetcherThrottleTimeSensor) { + this.applicationEventQueue = applicationEventQueue; + this.backgroundEventQueue = backgroundEventQueue; + final ConsumerMetadata metadata = bootstrapMetadata( + logContext, + clusterResourceListeners, + config, + subscriptionState + ); + final ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config, time, logContext); + final Selector selector = new Selector( + config.getLong( + ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), + metrics, + time, + METRIC_GRP_PREFIX, + channelBuilder, + logContext + ); + final NetworkClient netClient = new NetworkClient( + selector, + metadata, + config.getString(ConsumerConfig.CLIENT_ID_CONFIG), + 100, // a fixed large enough value will suffice for max + // in-flight requests + config.getLong(ConsumerConfig.RECONNECT_BACKOFF_MS_CONFIG), + config.getLong(ConsumerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG), + config.getInt(ConsumerConfig.SEND_BUFFER_CONFIG), + config.getInt(ConsumerConfig.RECEIVE_BUFFER_CONFIG), + config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG), + config.getLong(ConsumerConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG), + config.getLong(ConsumerConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG), + time, + true, + apiVersions, + fetcherThrottleTimeSensor, + logContext + ); + final ConsumerNetworkClient networkClient = new ConsumerNetworkClient( + logContext, + netClient, + metadata, + time, + config.getInt(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG), + config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG), + config.getInt(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG) + ); + this.backgroundThread = new DefaultBackgroundThread( + time, + config, + logContext, + this.applicationEventQueue, + this.backgroundEventQueue, + subscriptionState, + metadata, + networkClient, + new Metrics(time) + ); + } + + // VisibleForTesting + DefaultEventHandler(final Time time, + final ConsumerConfig config, + final LogContext logContext, + final BlockingQueue applicationEventQueue, + final BlockingQueue backgroundEventQueue, + final SubscriptionState subscriptionState, + final ConsumerMetadata metadata, + final ConsumerNetworkClient networkClient) { + this.applicationEventQueue = applicationEventQueue; + this.backgroundEventQueue = backgroundEventQueue; + this.backgroundThread = new DefaultBackgroundThread( + time, + config, + logContext, + this.applicationEventQueue, + this.backgroundEventQueue, + subscriptionState, + metadata, + networkClient, + new Metrics(time) + ); + backgroundThread.start(); + } + + // VisibleForTesting + DefaultEventHandler(final DefaultBackgroundThread backgroundThread, + final BlockingQueue applicationEventQueue, + final BlockingQueue backgroundEventQueue) { + this.backgroundThread = backgroundThread; + this.applicationEventQueue = applicationEventQueue; + this.backgroundEventQueue = backgroundEventQueue; + backgroundThread.start(); } @Override public Optional poll() { - return Optional.ofNullable(backgroundEvents.poll()); + return Optional.ofNullable(backgroundEventQueue.poll()); } @Override public boolean isEmpty() { - return backgroundEvents.isEmpty(); + return backgroundEventQueue.isEmpty(); } @Override - public boolean add(ApplicationEvent event) { + public boolean add(final ApplicationEvent event) { + backgroundThread.wakeup(); + return applicationEventQueue.add(event); + } + + // bootstrap a metadata object with the bootstrap server IP address, + // which will be used once for the subsequent metadata refresh once the + // background thread has started up. + private ConsumerMetadata bootstrapMetadata( + final LogContext logContext, + final ClusterResourceListeners clusterResourceListeners, + final ConsumerConfig config, + final SubscriptionState subscriptions) { + final ConsumerMetadata metadata = new ConsumerMetadata( + config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG), + config.getLong(ConsumerConfig.METADATA_MAX_AGE_CONFIG), + !config.getBoolean(ConsumerConfig.EXCLUDE_INTERNAL_TOPICS_CONFIG), + config.getBoolean(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG), + subscriptions, + logContext, clusterResourceListeners); + final List addresses = ClientUtils.parseAndValidateAddresses( + config.getList(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG), config.getString(ConsumerConfig.CLIENT_DNS_LOOKUP_CONFIG)); + metadata.bootstrap(addresses); + return metadata; + } + + public void close() { try { - return applicationEvents.add(event); - } catch (IllegalStateException e) { - // swallow the capacity restriction exception - return false; + backgroundThread.close(); + } catch (final Exception e) { + throw new RuntimeException(e); } } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NoopBackgroundEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NoopBackgroundEvent.java new file mode 100644 index 0000000000000..db6ed2f7a82dd --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NoopBackgroundEvent.java @@ -0,0 +1,36 @@ +/* + * 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. + */ +package org.apache.kafka.clients.consumer.internals; + +import org.apache.kafka.clients.consumer.internals.events.BackgroundEvent; + +/** + * Noop event. Intentionally left it here for demonstration purpose. + */ +public class NoopBackgroundEvent extends BackgroundEvent { + public final String message; + + public NoopBackgroundEvent(final String message) { + super(EventType.NOOP); + this.message = message; + } + + @Override + public String toString() { + return getClass() + "_" + this.message; + } +} diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java index 0764a27faf859..060fc3b50cdc1 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java @@ -49,6 +49,7 @@ public PrototypeAsyncConsumer(final Time time, final EventHandler eventHandler) * another type of event, process it. * 2. Send fetches if needed. * If the timeout expires, return an empty ConsumerRecord. + * * @param timeout timeout of the poll loop * @return ConsumerRecord. It can be empty if time timeout expires. */ @@ -57,7 +58,7 @@ public ConsumerRecords poll(final Duration timeout) { try { do { if (!eventHandler.isEmpty()) { - Optional backgroundEvent = eventHandler.poll(); + final Optional backgroundEvent = eventHandler.poll(); // processEvent() may process 3 types of event: // 1. Errors // 2. Callback Invocation @@ -76,7 +77,7 @@ public ConsumerRecords poll(final Duration timeout) { } // We will wait for retryBackoffMs } while (time.timer(timeout).notExpired()); - } catch (Exception e) { + } catch (final Exception e) { throw new RuntimeException(e); } @@ -84,7 +85,9 @@ public ConsumerRecords poll(final Duration timeout) { } abstract void processEvent(BackgroundEvent backgroundEvent, Duration timeout); + abstract ConsumerRecords processFetchResults(Fetch fetch); + abstract Fetch collectFetches(); /** @@ -92,25 +95,26 @@ public ConsumerRecords poll(final Duration timeout) { */ @Override public void commitAsync() { - ApplicationEvent commitEvent = new CommitApplicationEvent(); + final ApplicationEvent commitEvent = new CommitApplicationEvent(); eventHandler.add(commitEvent); } /** * This method sends a commit event to the EventHandler and waits for the event to finish. + * * @param timeout max wait time for the blocking operation. */ @Override - public void commitSync(Duration timeout) { - CommitApplicationEvent commitEvent = new CommitApplicationEvent(); + public void commitSync(final Duration timeout) { + final CommitApplicationEvent commitEvent = new CommitApplicationEvent(); eventHandler.add(commitEvent); - CompletableFuture commitFuture = commitEvent.commitFuture; + final CompletableFuture commitFuture = commitEvent.commitFuture; try { commitFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS); - } catch (TimeoutException e) { + } catch (final TimeoutException e) { throw new org.apache.kafka.common.errors.TimeoutException("timeout"); - } catch (Exception e) { + } catch (final Exception e) { // handle exception here throw new RuntimeException(e); } @@ -122,5 +126,13 @@ public void commitSync(Duration timeout) { private class CommitApplicationEvent extends ApplicationEvent { // this is the stubbed commitAsyncEvents CompletableFuture commitFuture = new CompletableFuture<>(); + + public CommitApplicationEvent() { + } + + @Override + public boolean process() { + return true; + } } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java index 0b2e5a901d5d2..683681a19fb14 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java @@ -20,4 +20,10 @@ * This is the abstract definition of the events created by the KafkaConsumer API */ abstract public class ApplicationEvent { + /** + * process the application event. Return true upon succesful execution, + * false otherwise. + * @return true if the event was successfully executed; false otherwise. + */ + public abstract boolean process(); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java index 8870e179d874e..89eac1048d95a 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEvent.java @@ -20,4 +20,12 @@ * This is the abstract definition of the events created by the background thread. */ abstract public class BackgroundEvent { + public final EventType type; + + public BackgroundEvent(EventType type) { + this.type = type; + } + public enum EventType { + NOOP, + } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/EventHandler.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/EventHandler.java index d9f5b9d065bbc..c07cc33cf7353 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/EventHandler.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/EventHandler.java @@ -14,7 +14,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.apache.kafka.clients.consumer.internals.events; import java.util.Optional; diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/NoopApplicationEvent.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/NoopApplicationEvent.java new file mode 100644 index 0000000000000..0fe9dccb1039a --- /dev/null +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/NoopApplicationEvent.java @@ -0,0 +1,46 @@ +/* + * 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. + */ +package org.apache.kafka.clients.consumer.internals.events; + +import org.apache.kafka.clients.consumer.internals.NoopBackgroundEvent; + +import java.util.concurrent.BlockingQueue; + +/** + * The event is NoOp. This is intentionally left here for demonstration purpose. + */ +public class NoopApplicationEvent extends ApplicationEvent { + public final String message; + private final BlockingQueue backgroundEventQueue; + + public NoopApplicationEvent(final BlockingQueue backgroundEventQueue, + final String message) { + + this.message = message; + this.backgroundEventQueue = backgroundEventQueue; + } + + @Override + public boolean process() { + return backgroundEventQueue.add(new NoopBackgroundEvent(message)); + } + + @Override + public String toString() { + return getClass() + "_" + this.message; + } +} diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java new file mode 100644 index 0000000000000..f159fad6bc521 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadTest.java @@ -0,0 +1,171 @@ +/* + * 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. + */ +package org.apache.kafka.clients.consumer.internals; + +import org.apache.kafka.clients.MockClient; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.internals.events.ApplicationEvent; +import org.apache.kafka.clients.consumer.internals.events.BackgroundEvent; +import org.apache.kafka.clients.consumer.internals.events.NoopApplicationEvent; +import org.apache.kafka.common.errors.WakeupException; +import org.apache.kafka.common.metrics.Metrics; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.common.utils.Timer; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.InOrder; +import org.mockito.Mockito; + +import java.util.Properties; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.RETRY_BACKOFF_MS_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class DefaultBackgroundThreadTest { + private static final long REFRESH_BACK_OFF_MS = 100; + private final Properties properties = new Properties(); + private MockTime time; + private SubscriptionState subscriptions; + private ConsumerMetadata metadata; + private LogContext context; + private ConsumerNetworkClient consumerClient; + private Metrics metrics; + private BlockingQueue backgroundEventsQueue; + private BlockingQueue applicationEventsQueue; + + @BeforeEach + @SuppressWarnings("unchecked") + public void setup() { + this.time = new MockTime(); + this.subscriptions = mock(SubscriptionState.class); + this.metadata = mock(ConsumerMetadata.class); + this.context = new LogContext(); + this.consumerClient = mock(ConsumerNetworkClient.class); + this.metrics = mock(Metrics.class); + this.applicationEventsQueue = (BlockingQueue) mock(BlockingQueue.class); + this.backgroundEventsQueue = (BlockingQueue) mock(BlockingQueue.class); + properties.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + properties.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + properties.put(RETRY_BACKOFF_MS_CONFIG, REFRESH_BACK_OFF_MS); + } + + @Test + public void testStartupAndTearDown() throws InterruptedException { + final MockClient client = new MockClient(time, metadata); + this.consumerClient = new ConsumerNetworkClient( + context, + client, + metadata, + time, + 100, + 1000, + 100 + ); + this.applicationEventsQueue = new LinkedBlockingQueue<>(); + final DefaultBackgroundThread backgroundThread = setupMockHandler(); + backgroundThread.start(); + assertTrue(client.active()); + backgroundThread.close(); + assertFalse(client.active()); + } + + @Test + public void testInterruption() throws InterruptedException { + final MockClient client = new MockClient(time, metadata); + this.consumerClient = new ConsumerNetworkClient( + context, + client, + metadata, + time, + 100, + 1000, + 100 + ); + this.applicationEventsQueue = new LinkedBlockingQueue<>(); + final DefaultBackgroundThread backgroundThread = setupMockHandler(); + backgroundThread.start(); + assertTrue(client.active()); + backgroundThread.close(); + assertFalse(client.active()); + } + + @Test + void testWakeup() { + this.time = new MockTime(0); + final MockClient client = new MockClient(time, metadata); + this.consumerClient = new ConsumerNetworkClient( + context, + client, + metadata, + time, + 100, + 1000, + 100 + ); + when(applicationEventsQueue.isEmpty()).thenReturn(true); + when(applicationEventsQueue.isEmpty()).thenReturn(true); + final DefaultBackgroundThread runnable = setupMockHandler(); + client.poll(0, time.milliseconds()); + runnable.wakeup(); + + assertThrows(WakeupException.class, runnable::runOnce); + runnable.close(); + } + + @Test + void testNetworkAndBlockingQueuePoll() { + // ensure network poll and application queue poll will happen in a + // single iteration + this.time = new MockTime(100); + final DefaultBackgroundThread runnable = setupMockHandler(); + runnable.runOnce(); + + when(applicationEventsQueue.isEmpty()).thenReturn(false); + when(applicationEventsQueue.poll()) + .thenReturn(new NoopApplicationEvent(backgroundEventsQueue, "nothing")); + final InOrder inOrder = Mockito.inOrder(applicationEventsQueue, this.consumerClient); + assertFalse(inOrder.verify(applicationEventsQueue).isEmpty()); + inOrder.verify(applicationEventsQueue).poll(); + inOrder.verify(this.consumerClient).poll(any(Timer.class)); + runnable.close(); + } + + private DefaultBackgroundThread setupMockHandler() { + return new DefaultBackgroundThread( + this.time, + new ConsumerConfig(properties), + new LogContext(), + applicationEventsQueue, + backgroundEventsQueue, + this.subscriptions, + this.metadata, + this.consumerClient, + this.metrics + ); + } +} diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandlerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandlerTest.java new file mode 100644 index 0000000000000..0da35e834945e --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandlerTest.java @@ -0,0 +1,116 @@ +/* + * 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. + */ +package org.apache.kafka.clients.consumer.internals; + +import org.apache.kafka.clients.MockClient; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.OffsetResetStrategy; +import org.apache.kafka.clients.consumer.internals.events.ApplicationEvent; +import org.apache.kafka.clients.consumer.internals.events.BackgroundEvent; +import org.apache.kafka.clients.consumer.internals.events.NoopApplicationEvent; +import org.apache.kafka.common.internals.ClusterResourceListeners; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.common.utils.Time; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import java.util.Optional; +import java.util.Properties; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + +import static org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.RETRY_BACKOFF_MS_CONFIG; +import static org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class DefaultEventHandlerTest { + private final Properties properties = new Properties(); + + @BeforeEach + public void setup() { + properties.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + properties.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + properties.put(RETRY_BACKOFF_MS_CONFIG, "100"); + } + + @Test + @Timeout(1) + public void testBasicPollAndAddWithNoopEvent() { + final Time time = new MockTime(1); + final LogContext logContext = new LogContext(); + final SubscriptionState subscriptions = new SubscriptionState(new LogContext(), OffsetResetStrategy.NONE); + final ConsumerMetadata metadata = newConsumerMetadata(false, subscriptions); + final MockClient client = new MockClient(time, metadata); + final ConsumerNetworkClient consumerClient = new ConsumerNetworkClient( + logContext, + client, + metadata, + time, + 100, + 1000, + 100 + ); + final BlockingQueue aq = new LinkedBlockingQueue<>(); + final BlockingQueue bq = new LinkedBlockingQueue<>(); + final DefaultEventHandler handler = new DefaultEventHandler( + time, + new ConsumerConfig(properties), + logContext, + aq, + bq, + subscriptions, + metadata, + consumerClient + ); + assertTrue(client.active()); + assertTrue(handler.isEmpty()); + handler.add( + new NoopApplicationEvent( + bq, + "testBasicPollAndAddWithNoopEvent" + ) + ); + while (handler.isEmpty()) { + time.sleep(100); + } + final Optional poll = handler.poll(); + assertTrue(poll.isPresent()); + assertTrue(poll.get() instanceof NoopBackgroundEvent); + + assertFalse(client.hasInFlightRequests()); // noop does not send network request + } + + private static ConsumerMetadata newConsumerMetadata(final boolean includeInternalTopics, + final SubscriptionState subscriptions) { + final long refreshBackoffMs = 50; + final long expireMs = 50000; + return new ConsumerMetadata( + refreshBackoffMs, + expireMs, + includeInternalTopics, + false, + subscriptions, + new LogContext(), + new ClusterResourceListeners() + ); + } +}