You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/10/07 22:34:08 UTC
[GitHub] [kafka] guozhangwang commented on a diff in pull request #12672: KAFKA-14247: Consumer background thread base implementation
guozhangwang commented on code in PR #12672:
URL: https://github.com/apache/kafka/pull/12672#discussion_r990452524
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/BackgroundThreadRunnable.java:
##########
@@ -0,0 +1,25 @@
+/*
+ * 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 java.io.Closeable;
+
+/**
+ * Background thread runnable that handles network IO such as fetching and committing.
+ */
+public interface BackgroundThreadRunnable extends Runnable, Closeable {
Review Comment:
I think the additional value of this interface is small given the already existed `DefaultEventHandler` class. That is, the default implementation of the event handler would always have a background thread anyways, while mocks would not have additional threads.
So I'd suggest we remove this interface and just let `DefaultEventHandler` contains a `BackgroundThread` directly.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadRunnable.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+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.NoopApplicationEvent;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.InterruptException;
+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 org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+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 DefaultBackgroundThreadRunnable implements BackgroundThreadRunnable {
+ private static final String METRIC_GRP_PREFIX = "consumer";
+
+ private final Time time;
+ private final Logger log;
+ private final BlockingQueue<ApplicationEvent> applicationEventQueue;
+ private final BlockingQueue<BackgroundEvent> 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<ApplicationEvent> inflightEvent = Optional.empty();
+ private AtomicReference<Optional<RuntimeException>> exception =
+ new AtomicReference<>(Optional.empty());
+
+ public DefaultBackgroundThreadRunnable(ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ this(Time.SYSTEM,
+ config,
+ logContext,
+ applicationEventQueue,
+ backgroundEventQueue,
+ subscriptions,
+ apiVersions,
+ metrics,
+ clusterResourceListeners,
+ fetcherThrottleTimeSensor);
+ }
+
+ public DefaultBackgroundThreadRunnable(Time time,
Review Comment:
How about extracting the construction of the `networkClient` out of the constructor of the thread, but in the constructor of the `DefaultEventHandler` and then pass in the `networkClient` into the thread? In that way we can save on passing many fields to its constructor.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadRunnable.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+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.NoopApplicationEvent;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.InterruptException;
+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 org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+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 DefaultBackgroundThreadRunnable implements BackgroundThreadRunnable {
+ private static final String METRIC_GRP_PREFIX = "consumer";
+
+ private final Time time;
+ private final Logger log;
+ private final BlockingQueue<ApplicationEvent> applicationEventQueue;
+ private final BlockingQueue<BackgroundEvent> 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<ApplicationEvent> inflightEvent = Optional.empty();
+ private AtomicReference<Optional<RuntimeException>> exception =
+ new AtomicReference<>(Optional.empty());
+
+ public DefaultBackgroundThreadRunnable(ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ this(Time.SYSTEM,
+ config,
+ logContext,
+ applicationEventQueue,
+ backgroundEventQueue,
+ subscriptions,
+ apiVersions,
+ metrics,
+ clusterResourceListeners,
+ fetcherThrottleTimeSensor);
+ }
+
+ public DefaultBackgroundThreadRunnable(Time time,
+ ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ try {
+ this.time = time;
+ this.log = logContext.logger(DefaultBackgroundThreadRunnable.class);
+ this.applicationEventQueue = applicationEventQueue;
+ this.backgroundEventQueue = backgroundEventQueue;
+ this.config = config;
+ setConfig();
+ this.inflightEvent = Optional.empty();
+ // subscriptionState is initialized in the polling thread
+ this.subscriptions = subscriptions;
+ this.metrics = metrics;
+ this.metadata = bootstrapMetadata(clusterResourceListeners, logContext);
+ ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config, time, logContext);
+ Selector selector = new Selector(config.getLong(
+ ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
+ metrics,
+ time,
+ METRIC_GRP_PREFIX,
+ channelBuilder,
+ logContext);
+ NetworkClient netClient = new NetworkClient(
+ selector,
+ metadata,
+ clientId,
+ 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);
+ this.networkClient = new ConsumerNetworkClient(
+ logContext,
+ netClient,
+ metadata,
+ time,
+ retryBackoffMs,
+ config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG),
+ heartbeatIntervalMs);
+ this.running = true;
+ } catch (Exception e) {
+ // now propagate the exception
+ close();
+ throw new KafkaException("Failed to construct background processor", e);
+ }
+ }
+
+ // VisibleForTesting
+ DefaultBackgroundThreadRunnable(Time time,
+ ConsumerConfig config,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ConsumerMetadata metadata,
+ ConsumerNetworkClient client) {
+ this.time = time;
+ this.config = config;
+ setConfig();
+ this.log = LoggerFactory.getLogger(getClass());
+ this.applicationEventQueue = applicationEventQueue;
+ this.backgroundEventQueue = backgroundEventQueue;
+ this.subscriptions = subscriptions;
+ this.metadata = metadata;
+ this.networkClient = client;
+ this.metrics = new Metrics();
+ this.running = true;
+ }
+
+ 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("{} started", getClass());
+ while (running) {
+ runOnce();
+ time.sleep(retryBackoffMs);
Review Comment:
Instead of setting a sleep here, I think it's better to have a `waitOnEmptyInputQueue` function with a condition. Where the caller thread would notify when enqueuing a new item.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadRunnable.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+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.NoopApplicationEvent;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.InterruptException;
+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 org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+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 DefaultBackgroundThreadRunnable implements BackgroundThreadRunnable {
+ private static final String METRIC_GRP_PREFIX = "consumer";
+
+ private final Time time;
+ private final Logger log;
+ private final BlockingQueue<ApplicationEvent> applicationEventQueue;
+ private final BlockingQueue<BackgroundEvent> 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<ApplicationEvent> inflightEvent = Optional.empty();
+ private AtomicReference<Optional<RuntimeException>> exception =
+ new AtomicReference<>(Optional.empty());
+
+ public DefaultBackgroundThreadRunnable(ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ this(Time.SYSTEM,
+ config,
+ logContext,
+ applicationEventQueue,
+ backgroundEventQueue,
+ subscriptions,
+ apiVersions,
+ metrics,
+ clusterResourceListeners,
+ fetcherThrottleTimeSensor);
+ }
+
+ public DefaultBackgroundThreadRunnable(Time time,
+ ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ try {
+ this.time = time;
+ this.log = logContext.logger(DefaultBackgroundThreadRunnable.class);
+ this.applicationEventQueue = applicationEventQueue;
+ this.backgroundEventQueue = backgroundEventQueue;
+ this.config = config;
+ setConfig();
+ this.inflightEvent = Optional.empty();
+ // subscriptionState is initialized in the polling thread
+ this.subscriptions = subscriptions;
+ this.metrics = metrics;
+ this.metadata = bootstrapMetadata(clusterResourceListeners, logContext);
+ ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config, time, logContext);
+ Selector selector = new Selector(config.getLong(
+ ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
+ metrics,
+ time,
+ METRIC_GRP_PREFIX,
+ channelBuilder,
+ logContext);
+ NetworkClient netClient = new NetworkClient(
+ selector,
+ metadata,
+ clientId,
+ 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);
+ this.networkClient = new ConsumerNetworkClient(
+ logContext,
+ netClient,
+ metadata,
+ time,
+ retryBackoffMs,
+ config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG),
+ heartbeatIntervalMs);
+ this.running = true;
+ } catch (Exception e) {
+ // now propagate the exception
+ close();
+ throw new KafkaException("Failed to construct background processor", e);
+ }
+ }
+
+ // VisibleForTesting
+ DefaultBackgroundThreadRunnable(Time time,
+ ConsumerConfig config,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ConsumerMetadata metadata,
+ ConsumerNetworkClient client) {
+ this.time = time;
+ this.config = config;
+ setConfig();
+ this.log = LoggerFactory.getLogger(getClass());
+ this.applicationEventQueue = applicationEventQueue;
+ this.backgroundEventQueue = backgroundEventQueue;
+ this.subscriptions = subscriptions;
+ this.metadata = metadata;
+ this.networkClient = client;
+ this.metrics = new Metrics();
+ this.running = true;
+ }
+
+ 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("{} started", getClass());
+ while (running) {
+ runOnce();
+ time.sleep(retryBackoffMs);
+ }
+ } catch (InterruptException e) {
+ log.error("The background thread has been interrupted");
+ this.exception.set(Optional.of(new RuntimeException(e)));
+ } catch (Throwable t) {
+ log.error("The background 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();
+ log.debug("processing applicatoin event: {}", this.inflightEvent);
+ if (this.inflightEvent.isPresent() && maybeConsumeInflightEvent(this.inflightEvent.get())) {
+ // clear inflight event upon successful consumption
+ this.inflightEvent = Optional.empty();
+ }
+ networkClient.pollNoWakeup();
+ }
+
+ public Optional<ApplicationEvent> 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.
+ */
+ public boolean maybeConsumeInflightEvent(ApplicationEvent event) {
+ log.debug("try consuming event: {}", Optional.ofNullable(event));
+ switch (event.type) {
+ case NOOP:
+ process((NoopApplicationEvent) event);
+ return true;
+ default:
+ inflightEvent = Optional.empty();
+ log.warn("unsupported event type: {}", event.type);
+ return true;
+ }
+ }
+
+ /**
+ * Processes {@link NoopApplicationEvent} and equeue a {@link NoopBackgroundEvent}. This is intentionally left here
+ * for demonstration purpose.
+ * @param event a {@link NoopApplicationEvent}
+ */
+ private void process(NoopApplicationEvent event) {
+ backgroundEventQueue.add(new NoopBackgroundEvent(event.message));
+ }
+
+ public boolean isRunning() {
+ return this.running;
+ }
+
+ public Optional<RuntimeException> exception() {
Review Comment:
This one seems not used?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadRunnable.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+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.NoopApplicationEvent;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.InterruptException;
+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 org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+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 DefaultBackgroundThreadRunnable implements BackgroundThreadRunnable {
+ private static final String METRIC_GRP_PREFIX = "consumer";
+
+ private final Time time;
+ private final Logger log;
+ private final BlockingQueue<ApplicationEvent> applicationEventQueue;
+ private final BlockingQueue<BackgroundEvent> 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<ApplicationEvent> inflightEvent = Optional.empty();
+ private AtomicReference<Optional<RuntimeException>> exception =
+ new AtomicReference<>(Optional.empty());
+
+ public DefaultBackgroundThreadRunnable(ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ this(Time.SYSTEM,
+ config,
+ logContext,
+ applicationEventQueue,
+ backgroundEventQueue,
+ subscriptions,
+ apiVersions,
+ metrics,
+ clusterResourceListeners,
+ fetcherThrottleTimeSensor);
+ }
+
+ public DefaultBackgroundThreadRunnable(Time time,
+ ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ try {
+ this.time = time;
+ this.log = logContext.logger(DefaultBackgroundThreadRunnable.class);
+ this.applicationEventQueue = applicationEventQueue;
+ this.backgroundEventQueue = backgroundEventQueue;
+ this.config = config;
+ setConfig();
+ this.inflightEvent = Optional.empty();
+ // subscriptionState is initialized in the polling thread
+ this.subscriptions = subscriptions;
+ this.metrics = metrics;
+ this.metadata = bootstrapMetadata(clusterResourceListeners, logContext);
+ ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config, time, logContext);
+ Selector selector = new Selector(config.getLong(
+ ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
+ metrics,
+ time,
+ METRIC_GRP_PREFIX,
+ channelBuilder,
+ logContext);
+ NetworkClient netClient = new NetworkClient(
+ selector,
+ metadata,
+ clientId,
+ 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);
+ this.networkClient = new ConsumerNetworkClient(
+ logContext,
+ netClient,
+ metadata,
+ time,
+ retryBackoffMs,
+ config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG),
+ heartbeatIntervalMs);
+ this.running = true;
+ } catch (Exception e) {
+ // now propagate the exception
+ close();
+ throw new KafkaException("Failed to construct background processor", e);
+ }
+ }
+
+ // VisibleForTesting
+ DefaultBackgroundThreadRunnable(Time time,
+ ConsumerConfig config,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ConsumerMetadata metadata,
+ ConsumerNetworkClient client) {
+ this.time = time;
+ this.config = config;
+ setConfig();
+ this.log = LoggerFactory.getLogger(getClass());
+ this.applicationEventQueue = applicationEventQueue;
+ this.backgroundEventQueue = backgroundEventQueue;
+ this.subscriptions = subscriptions;
+ this.metadata = metadata;
+ this.networkClient = client;
+ this.metrics = new Metrics();
+ this.running = true;
+ }
+
+ 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("{} started", getClass());
+ while (running) {
+ runOnce();
+ time.sleep(retryBackoffMs);
+ }
+ } catch (InterruptException e) {
+ log.error("The background thread has been interrupted");
+ this.exception.set(Optional.of(new RuntimeException(e)));
+ } catch (Throwable t) {
+ log.error("The background 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();
+ log.debug("processing applicatoin event: {}", this.inflightEvent);
+ if (this.inflightEvent.isPresent() && maybeConsumeInflightEvent(this.inflightEvent.get())) {
+ // clear inflight event upon successful consumption
+ this.inflightEvent = Optional.empty();
+ }
+ networkClient.pollNoWakeup();
+ }
+
+ public Optional<ApplicationEvent> 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.
+ */
+ public boolean maybeConsumeInflightEvent(ApplicationEvent event) {
+ log.debug("try consuming event: {}", Optional.ofNullable(event));
+ switch (event.type) {
+ case NOOP:
+ process((NoopApplicationEvent) event);
+ return true;
+ default:
+ inflightEvent = Optional.empty();
+ log.warn("unsupported event type: {}", event.type);
+ return true;
+ }
+ }
+
+ /**
+ * Processes {@link NoopApplicationEvent} and equeue a {@link NoopBackgroundEvent}. This is intentionally left here
+ * for demonstration purpose.
+ * @param event a {@link NoopApplicationEvent}
+ */
+ private void process(NoopApplicationEvent event) {
Review Comment:
This seems like what a mock client needs to do, why we need to add it to the default implementation? Or are we going to remove it in this class later?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandlerTest.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.EventHandler;
+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.assertEquals;
+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() {
+ Time time = new MockTime(1);
+ LogContext logContext = new LogContext();
+ SubscriptionState subscriptions = new SubscriptionState(new LogContext(), OffsetResetStrategy.NONE);
+ ConsumerMetadata metadata = newConsumerMetadata(false, subscriptions);
+ MockClient client = new MockClient(time, metadata);
+ ConsumerNetworkClient consumerClient = new ConsumerNetworkClient(logContext, client, metadata, time,
+ 100, 1000, 100);
+ BlockingQueue<ApplicationEvent> aq = new LinkedBlockingQueue<>();
+ BlockingQueue<BackgroundEvent> bq = new LinkedBlockingQueue<>();
+ DefaultEventHandler handler = new DefaultEventHandler(
+ time,
+ new ConsumerConfig(properties),
+ aq, bq,
Review Comment:
nit: newline for new param.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/NoopApplicationEvent.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * 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;
+
+/**
+ * The event is NoOp. This is intentionally left here for demonstration purpose.
+ */
+public class NoopApplicationEvent extends ApplicationEvent {
Review Comment:
I think we should move this class to the test package instead of the main package to separate it from other production, non-testing code.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandlerTest.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.EventHandler;
+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.assertEquals;
+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() {
+ Time time = new MockTime(1);
+ LogContext logContext = new LogContext();
+ SubscriptionState subscriptions = new SubscriptionState(new LogContext(), OffsetResetStrategy.NONE);
+ ConsumerMetadata metadata = newConsumerMetadata(false, subscriptions);
+ MockClient client = new MockClient(time, metadata);
+ ConsumerNetworkClient consumerClient = new ConsumerNetworkClient(logContext, client, metadata, time,
+ 100, 1000, 100);
+ BlockingQueue<ApplicationEvent> aq = new LinkedBlockingQueue<>();
+ BlockingQueue<BackgroundEvent> bq = new LinkedBlockingQueue<>();
+ DefaultEventHandler handler = new DefaultEventHandler(
+ time,
+ new ConsumerConfig(properties),
+ aq, bq,
+ subscriptions,
+ metadata,
+ consumerClient);
+ assertTrue(client.active());
+ assertTrue(handler.isEmpty());
+ handler.add(new NoopApplicationEvent("testBasicPollAndAddWithNoopEvent"));
+ while (handler.isEmpty()) {
+ time.sleep(100);
+ }
+ assertTrue(handler.poll().get() instanceof NoopBackgroundEvent);
+ assertFalse(client.hasInFlightRequests()); // noop does not send network request
+ handler.close();
+ assertFalse(client.active());
+ }
+
+ @Test
+ @Timeout(1)
+ public void testRunOnceBackgroundThread() {
+ BlockingQueue<ApplicationEvent> applicationEventQueue = new LinkedBlockingQueue<>();
+ BlockingQueue<BackgroundEvent> backgroundEventQueue = new LinkedBlockingQueue<>();
+ EventHandler eventHandler = new DefaultEventHandler(
+ new RunOnceBackgroundThreadRunnable(applicationEventQueue, backgroundEventQueue),
Review Comment:
It's unclear to me why we would need a separate background thread to test this functionality? I think just using the default thread would be sufficient still?
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandlerTest.java:
##########
@@ -0,0 +1,135 @@
+/*
+ * 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.EventHandler;
+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.assertEquals;
+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() {
+ Time time = new MockTime(1);
+ LogContext logContext = new LogContext();
+ SubscriptionState subscriptions = new SubscriptionState(new LogContext(), OffsetResetStrategy.NONE);
+ ConsumerMetadata metadata = newConsumerMetadata(false, subscriptions);
+ MockClient client = new MockClient(time, metadata);
+ ConsumerNetworkClient consumerClient = new ConsumerNetworkClient(logContext, client, metadata, time,
+ 100, 1000, 100);
+ BlockingQueue<ApplicationEvent> aq = new LinkedBlockingQueue<>();
+ BlockingQueue<BackgroundEvent> bq = new LinkedBlockingQueue<>();
+ DefaultEventHandler handler = new DefaultEventHandler(
+ time,
+ new ConsumerConfig(properties),
+ aq, bq,
+ subscriptions,
+ metadata,
+ consumerClient);
+ assertTrue(client.active());
+ assertTrue(handler.isEmpty());
+ handler.add(new NoopApplicationEvent("testBasicPollAndAddWithNoopEvent"));
+ while (handler.isEmpty()) {
+ time.sleep(100);
+ }
+ assertTrue(handler.poll().get() instanceof NoopBackgroundEvent);
+ assertFalse(client.hasInFlightRequests()); // noop does not send network request
+ handler.close();
+ assertFalse(client.active());
+ }
+
+ @Test
+ @Timeout(1)
+ public void testRunOnceBackgroundThread() {
+ BlockingQueue<ApplicationEvent> applicationEventQueue = new LinkedBlockingQueue<>();
+ BlockingQueue<BackgroundEvent> backgroundEventQueue = new LinkedBlockingQueue<>();
+ EventHandler eventHandler = new DefaultEventHandler(
+ new RunOnceBackgroundThreadRunnable(applicationEventQueue, backgroundEventQueue),
+ applicationEventQueue,
+ backgroundEventQueue);
+ assertTrue(eventHandler.add(new NoopApplicationEvent("hello-world")));
+ while (eventHandler.isEmpty()) { }
Review Comment:
This is a general comment: I think it worth adding a `poll(timeout)` in the `EventHandler` interface since 1) for testing, we do not need the while loop any more, and 2) in production code, for the caller thread scenarios I think there are cases where a `poll(timeout)` would be useful as well.
##########
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 {
Review Comment:
Ditto here, I think we should move it to test package, not main package.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadRunnable.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+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.NoopApplicationEvent;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.InterruptException;
+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 org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+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 DefaultBackgroundThreadRunnable implements BackgroundThreadRunnable {
+ private static final String METRIC_GRP_PREFIX = "consumer";
+
+ private final Time time;
+ private final Logger log;
+ private final BlockingQueue<ApplicationEvent> applicationEventQueue;
+ private final BlockingQueue<BackgroundEvent> 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<ApplicationEvent> inflightEvent = Optional.empty();
+ private AtomicReference<Optional<RuntimeException>> exception =
+ new AtomicReference<>(Optional.empty());
+
+ public DefaultBackgroundThreadRunnable(ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ this(Time.SYSTEM,
+ config,
+ logContext,
+ applicationEventQueue,
+ backgroundEventQueue,
+ subscriptions,
+ apiVersions,
+ metrics,
+ clusterResourceListeners,
+ fetcherThrottleTimeSensor);
+ }
+
+ public DefaultBackgroundThreadRunnable(Time time,
Review Comment:
And in this way we would not need two constructors either.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadRunnable.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+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.NoopApplicationEvent;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.InterruptException;
+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 org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+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 DefaultBackgroundThreadRunnable implements BackgroundThreadRunnable {
+ private static final String METRIC_GRP_PREFIX = "consumer";
+
+ private final Time time;
+ private final Logger log;
+ private final BlockingQueue<ApplicationEvent> applicationEventQueue;
+ private final BlockingQueue<BackgroundEvent> 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<ApplicationEvent> inflightEvent = Optional.empty();
+ private AtomicReference<Optional<RuntimeException>> exception =
+ new AtomicReference<>(Optional.empty());
+
+ public DefaultBackgroundThreadRunnable(ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ this(Time.SYSTEM,
+ config,
+ logContext,
+ applicationEventQueue,
+ backgroundEventQueue,
+ subscriptions,
+ apiVersions,
+ metrics,
+ clusterResourceListeners,
+ fetcherThrottleTimeSensor);
+ }
+
+ public DefaultBackgroundThreadRunnable(Time time,
Review Comment:
Ditto for `ConsumerMetadata`.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadRunnable.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+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.common.KafkaException;
+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 org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * Lives inside of the {@code DefaultEventHandler}, and consumes {@code ApplicationEvent} and produces
+ * {@code BackgroundEvent}.
+ */
+public class DefaultBackgroundThreadRunnable implements BackgroundThreadRunnable {
+ private static final String CLIENT_ID_METRIC_TAG = "client-id";
+ private static final String METRIC_GRP_PREFIX = "consumer";
+
+ private final Time time;
+ private final Logger log;
+ private final BlockingQueue<ApplicationEvent> applicationEventQueue;
+ private final BlockingQueue<BackgroundEvent> backgroundEventQueue;
+ private final ConsumerNetworkClient networkClient;
+
+ private final ConsumerConfig config;
+ private final Metrics metrics;
+ private final SubscriptionState subscriptions;
+ private final ConsumerMetadata metadata;
+ private String clientId;
+ private long retryBackoffMs;
+ private int heartbeatIntervalMs;
+ private boolean running = false;
+ private Optional<ApplicationEvent> inflightEvent;
+
+ public DefaultBackgroundThreadRunnable(ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ this(Time.SYSTEM,
+ config,
+ logContext,
+ applicationEventQueue,
+ backgroundEventQueue,
+ subscriptions,
+ apiVersions,
+ metrics,
+ clusterResourceListeners,
+ fetcherThrottleTimeSensor);
+ }
+
+ public DefaultBackgroundThreadRunnable(Time time,
+ ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ try {
+ this.time = time;
+ this.log = logContext.logger(DefaultBackgroundThreadRunnable.class);
+ this.applicationEventQueue = applicationEventQueue;
+ this.backgroundEventQueue = backgroundEventQueue;
+ this.config = config;
+ setConfig();
+ this.inflightEvent = Optional.empty();
+ this.subscriptions = subscriptions;
+ this.metrics = metrics;
+ this.metadata = bootstrapMetadata(clusterResourceListeners, logContext);
+ ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config, time, logContext);
+ NetworkClient netClient = new NetworkClient(
+ new Selector(config.getLong(ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG), metrics, time, METRIC_GRP_PREFIX, channelBuilder, logContext),
+ this.metadata,
+ clientId,
+ 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);
+ this.networkClient = new ConsumerNetworkClient(
+ logContext,
+ netClient,
+ metadata,
+ time,
+ retryBackoffMs,
+ config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG),
+ heartbeatIntervalMs);
+ } catch (Exception e) {
+ // now propagate the exception
+ throw new KafkaException("Failed to construct background processor", e);
+ }
+ }
+
+ // VisibleForTesting
+ DefaultBackgroundThreadRunnable(Time time,
+ ConsumerConfig config,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ConsumerMetadata metadata,
+ ConsumerNetworkClient client) {
+ this.time = time;
+ this.config = config;
+ this.log = LoggerFactory.getLogger(getClass());
+ this.applicationEventQueue = applicationEventQueue;
+ this.backgroundEventQueue = backgroundEventQueue;
+ this.subscriptions = subscriptions;
+ this.metadata = metadata;
+ this.networkClient = client;
+ this.metrics = new Metrics();
+ }
+
+ 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("DefaultBackgroundThreadRunnable started");
+ while (running) {
+ pollOnce();
+ this.wait(retryBackoffMs);
+ }
+ } catch (Exception e) {
+ // TODO: Define fine grain exceptions here
+ } finally {
+ log.debug("DefaultBackgroundThreadRunnable closed");
+ }
+ }
+
+ /**
+ * Process event from a single poll
+ */
+ void pollOnce() {
+ this.inflightEvent = maybePollEvent();
+ if (this.inflightEvent.isPresent() && maybeConsumeInflightEvent(this.inflightEvent.get())) {
+ // clear inflight event upon successful consumption
+ this.inflightEvent = Optional.empty();
+ }
+ networkClient.pollNoWakeup();
+ }
+
+ public Optional<ApplicationEvent> maybePollEvent() {
+ if (this.inflightEvent.isPresent()) {
+ return this.inflightEvent;
+ }
+ return Optional.ofNullable(this.applicationEventQueue.poll());
Review Comment:
+1 to @philipnee 's comment. The main reason is that I think eventually the background thread would not only try to just take events from the queue, since even though there's no events incoming, it would still try to repeatedly send fetch request and handle fetch responses -- well, for now maybe we cannot accumulate more than one fetched response per partition, but we may lift that with a memory limit in the future --- so the `runOnce` pattern would likely be:
1. try to get some event from the queue, if yes, then handle it and potentially send the corresponding request;
2. if the queue is empty, then try to send a fetch request.
3. poll the network client in order to receive any responses ready-to-read on the socket.
In that case, we would not want to wait on the queue at all if there's none, but directly move on to step 2) in each iteration.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadRunnable.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+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.NoopApplicationEvent;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.InterruptException;
+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 org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+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 DefaultBackgroundThreadRunnable implements BackgroundThreadRunnable {
Review Comment:
It's probably better to extend the `Thread` class (which implements Runnable anyways) for a couple of reasons: 1) we can set a meaningful thread name, which would be useful when we add metrics for the background activities, as part of the metric tags. 2) The shutting down procedure may be cleaner with the built-in `shutdown and join` mechanisms with the `Thread` class.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandler.java:
##########
@@ -16,45 +16,114 @@
*/
package org.apache.kafka.clients.consumer.internals;
+import org.apache.kafka.clients.ApiVersions;
+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.utils.KafkaThread;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
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<ApplicationEvent> applicationEvents;
- private final BlockingQueue<BackgroundEvent> backgroundEvents;
+ private final BlockingQueue<ApplicationEvent> applicationEventQueue;
+ private final BlockingQueue<BackgroundEvent> backgroundEventQueue;
+ private final BackgroundThreadRunnable runnable;
+ private final KafkaThread 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(ConsumerConfig config, LogContext logcontext,
+ SubscriptionState subscriptionState,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor,
+ ApiVersions apiVersions) {
+ this.applicationEventQueue = new LinkedBlockingQueue<>();
+ this.backgroundEventQueue = new LinkedBlockingQueue<>();
+ this.runnable = new DefaultBackgroundThreadRunnable(
+ config,
+ logcontext,
+ applicationEventQueue,
+ backgroundEventQueue,
+ subscriptionState,
+ apiVersions,
+ metrics,
+ clusterResourceListeners,
+ fetcherThrottleTimeSensor);
+ this.backgroundThread = new KafkaThread("consumer_background_thread", runnable, true);
Review Comment:
Related to my other comment in the `DefaultBackgroundThreadRunnable`: I think we can let that class to extend `Thread` directly (and then we can also rename it to `BackgroundThread` or `ApplicationEventHandlerThread`)?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadRunnable.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.ApiVersions;
+import org.apache.kafka.clients.ClientUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
+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.NoopApplicationEvent;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.InterruptException;
+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 org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+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 DefaultBackgroundThreadRunnable implements BackgroundThreadRunnable {
+ private static final String METRIC_GRP_PREFIX = "consumer";
+
+ private final Time time;
+ private final Logger log;
+ private final BlockingQueue<ApplicationEvent> applicationEventQueue;
+ private final BlockingQueue<BackgroundEvent> 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<ApplicationEvent> inflightEvent = Optional.empty();
+ private AtomicReference<Optional<RuntimeException>> exception =
+ new AtomicReference<>(Optional.empty());
+
+ public DefaultBackgroundThreadRunnable(ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ this(Time.SYSTEM,
+ config,
+ logContext,
+ applicationEventQueue,
+ backgroundEventQueue,
+ subscriptions,
+ apiVersions,
+ metrics,
+ clusterResourceListeners,
+ fetcherThrottleTimeSensor);
+ }
+
+ public DefaultBackgroundThreadRunnable(Time time,
+ ConsumerConfig config,
+ LogContext logContext,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ApiVersions apiVersions,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor) {
+ try {
+ this.time = time;
+ this.log = logContext.logger(DefaultBackgroundThreadRunnable.class);
+ this.applicationEventQueue = applicationEventQueue;
+ this.backgroundEventQueue = backgroundEventQueue;
+ this.config = config;
+ setConfig();
+ this.inflightEvent = Optional.empty();
+ // subscriptionState is initialized in the polling thread
+ this.subscriptions = subscriptions;
+ this.metrics = metrics;
+ this.metadata = bootstrapMetadata(clusterResourceListeners, logContext);
+ ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config, time, logContext);
+ Selector selector = new Selector(config.getLong(
+ ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
+ metrics,
+ time,
+ METRIC_GRP_PREFIX,
+ channelBuilder,
+ logContext);
+ NetworkClient netClient = new NetworkClient(
+ selector,
+ metadata,
+ clientId,
+ 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);
+ this.networkClient = new ConsumerNetworkClient(
+ logContext,
+ netClient,
+ metadata,
+ time,
+ retryBackoffMs,
+ config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG),
+ heartbeatIntervalMs);
+ this.running = true;
+ } catch (Exception e) {
+ // now propagate the exception
+ close();
+ throw new KafkaException("Failed to construct background processor", e);
+ }
+ }
+
+ // VisibleForTesting
+ DefaultBackgroundThreadRunnable(Time time,
+ ConsumerConfig config,
+ BlockingQueue<ApplicationEvent> applicationEventQueue,
+ BlockingQueue<BackgroundEvent> backgroundEventQueue,
+ SubscriptionState subscriptions,
+ ConsumerMetadata metadata,
+ ConsumerNetworkClient client) {
+ this.time = time;
+ this.config = config;
+ setConfig();
+ this.log = LoggerFactory.getLogger(getClass());
+ this.applicationEventQueue = applicationEventQueue;
+ this.backgroundEventQueue = backgroundEventQueue;
+ this.subscriptions = subscriptions;
+ this.metadata = metadata;
+ this.networkClient = client;
+ this.metrics = new Metrics();
+ this.running = true;
+ }
+
+ 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("{} started", getClass());
+ while (running) {
+ runOnce();
+ time.sleep(retryBackoffMs);
+ }
+ } catch (InterruptException e) {
+ log.error("The background thread has been interrupted");
Review Comment:
Why we want to retain interrupt exceptions as well, instead of just ignoring it and continue the next iteration?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandler.java:
##########
@@ -16,45 +16,114 @@
*/
package org.apache.kafka.clients.consumer.internals;
+import org.apache.kafka.clients.ApiVersions;
+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.utils.KafkaThread;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.Time;
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<ApplicationEvent> applicationEvents;
- private final BlockingQueue<BackgroundEvent> backgroundEvents;
+ private final BlockingQueue<ApplicationEvent> applicationEventQueue;
+ private final BlockingQueue<BackgroundEvent> backgroundEventQueue;
+ private final BackgroundThreadRunnable runnable;
+ private final KafkaThread 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(ConsumerConfig config, LogContext logcontext,
+ SubscriptionState subscriptionState,
+ Metrics metrics,
+ ClusterResourceListeners clusterResourceListeners,
+ Sensor fetcherThrottleTimeSensor,
+ ApiVersions apiVersions) {
+ this.applicationEventQueue = new LinkedBlockingQueue<>();
+ this.backgroundEventQueue = new LinkedBlockingQueue<>();
+ this.runnable = new DefaultBackgroundThreadRunnable(
+ config,
+ logcontext,
+ applicationEventQueue,
+ backgroundEventQueue,
+ subscriptionState,
+ apiVersions,
+ metrics,
+ clusterResourceListeners,
+ fetcherThrottleTimeSensor);
+ this.backgroundThread = new KafkaThread("consumer_background_thread", runnable, true);
+ backgroundThread.start();
+
+ }
+
+ // VisibleForTesting
+ DefaultEventHandler(BackgroundThreadRunnable runnable,
Review Comment:
I had a similar thought about those testing-used constructors as the background thread class: for this one, it seems to me that just using the default thread is also okay (see my other comment in the test class); for the one in line 80, I think it's better to let it's caller, a.k.a. KafkaConsumer to construct the aq and bq and then pass them into the constructors, in which case we only need that constructor, and not the one in line 45 any more.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEvent.java:
##########
@@ -20,4 +20,14 @@
* This is the abstract definition of the events created by the KafkaConsumer API
*/
abstract public class ApplicationEvent {
+ public final EventType type;
+
+ public ApplicationEvent(EventType type) {
+ this.type = type;
+ }
+
+ public enum EventType {
+ COMMIT,
+ NOOP,
Review Comment:
Ditto, I think we should keep the testing-related element out of the production code eventually. If we are going to remove it later then that's fine.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThreadRunnableTest.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.common.internals.ClusterResourceListeners;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.utils.KafkaThread;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.common.utils.MockTime;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Properties;
+import java.util.concurrent.LinkedBlockingDeque;
+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.VALUE_DESERIALIZER_CLASS_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class DefaultBackgroundThreadRunnableTest {
+ private long refreshBackoffMs = 100;
+ private long expireMs = 1000;
+ private final Properties properties = new Properties();
+ private MockTime time;
+ private SubscriptionState subscriptions;
+ private ConsumerMetadata metadata;
+ private MockClient client;
+ private LogContext context;
+ private ConsumerNetworkClient consumerClient;
+
+ @BeforeEach
+ public void setup() {
+ this.time = new MockTime();
+ this.subscriptions = new SubscriptionState(new LogContext(), OffsetResetStrategy.NONE);
+ this.metadata = new ConsumerMetadata(refreshBackoffMs, expireMs, false, false,
+ subscriptions, new LogContext(), new ClusterResourceListeners());
+ this.client = new MockClient(time, metadata);
+ this.context = new LogContext();
+ this.consumerClient = new ConsumerNetworkClient(context, client, metadata, time,
+ 100, 1000, 100);
+ properties.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ properties.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ }
+
+ @Test
+ public void testStartupAndTearDown() {
+ DefaultBackgroundThreadRunnable runnable = setupMockHandler();
+ KafkaThread thread = new KafkaThread("test-thread", runnable, true);
+ thread.start();
+ assertTrue(client.active());
+ runnable.close();
+ assertFalse(client.active());
+ }
+
+ private DefaultBackgroundThreadRunnable setupMockHandler() {
Review Comment:
Could we add test coverage here for: 1) it will poll network client (We can mock the network client) in each iteration; 2) it will try to poll from the aq in each iteration.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org