KAFKA-14274 [6, 7]: Introduction of fetch request manager by kirktrue · Pull Request #14406 · apache/kafka

kirktrue · 2023-09-19T17:26:32Z

Changes:

Introduces FetchRequestManager that implements the RequestManager API for fetching messages from brokers. Unlike Fetcher, record decompression and deserialization is performed on the application thread inside CompletedFetch.
Restructured the code so that objects owned by the background thread are not instantiated until the background thread runs (via Supplier) to ensure that there are no references available to the application thread.
Ensuring resources are properly using Closeable and using IdempotentCloser to ensure they're only closed once.
Introduces ConsumerTestBuilder to reduce a lot of inconsistency in the way the objects were built up for tests.

Changes: 1. Introduces FetchRequestManager that implements the RequestManager API for fetching messages from brokers. Unlike Fetcher, record decompression and deserialization is performed on the application thread inside CompletedFetch. 2. Restructured the code so that objects owned by the background thread are not instantiated until the background thread runs (via Supplier) to ensure that there are no references available to the application thread. 3. Ensuring resources are properly using Closeable and using IdempotentCloser to ensure they're only closed once. 4. Introduces ConsumerTestBuilder to reduce a lot of inconsistency in the way the objects were built up for tests.

…rializers

kirktrue · 2023-09-20T21:19:06Z

@philipnee Can you add the ctr label, please 🥺

checkstyle/import-control.xml

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ErrorEventHandler.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java

...c/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEventProcessor.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandler.java

junrao

@kirktrue : Thanks for the PR. Made a pass of non-testing files. Left a few comments.

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchConfig.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManager.java

clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManagers.java

Changes: 1. Introduces FetchRequestManager that implements the RequestManager API for fetching messages from brokers. Unlike Fetcher, record decompression and deserialization is performed on the application thread inside CompletedFetch. 2. Restructured the code so that objects owned by the background thread are not instantiated until the background thread runs (via Supplier) to ensure that there are no references available to the application thread. 3. Ensuring resources are properly using Closeable and using IdempotentCloser to ensure they're only closed once. 4. Introduces ConsumerTestBuilder to reduce a lot of inconsistency in the way the objects were built up for tests.

…rializers

…f github.com:kirktrue/kafka into KAFKA-14274-3-introduce-fetch-request-manager-take-2

Changes: 1. Introduced a new generic queue processing class that both the ApplicationEventProcessor and the BackgroundEventProcessor use. 2. Moved event processing to each iteration of the internal loop inside PrototypeAsyncConsumer.poll, instead of just at the start. 3. Removed the NOOP event types. 4. Introduced the CompletableEvent interface which will be used by the BackgroundEvent hierarchy soon 5. Provided missing documentation around the queue processors. 6. No longer returning a flag from the event processor method as it was not used anywhere.

…er-take-2

Refactored the close method so that it's much cleaner. Also fixed a confusing doc change to refreshCommittedOffsetsIfNeeded().

FetchEvent now provides the Future to the FetchRequestManager which will directly respond with the results of the fetch. This allows the application thread to properly block in pollForFetches() as it waits for a response. A separate thread-safe blocking queue collects the results from the Future, as opposed to collecting the results from the future in the fetch buffer, as that is not thread safe and should not be used across threads.

1. Renamed DefaultEventHandler to ApplicationEventHandler 2. Renamed ErrorEventHandler to BackgroundEventHandler 3. Deleted EventHandler interface 4. Moved ApplicationEventHandler and BackgroundEventHandler (and their unit tests) into the "events" sub-package for consistency 5. Removed use of mocks in DefaultBackgroundThreadTest as they caused issues with various methods returning null

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java

…dundancy

…er-take-2

junrao

@kirktrue : Thanks for the updated PR. Added a few more comments.

junrao · 2023-10-20T22:03:20Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

+
+        if (!fetch.isEmpty()) {
+            // Notify the network thread to wake up and start the next round of fetching.
+            applicationEventHandler.wakeup();


This doesn't seem quite right. It's possible for a fetchRequest to return no data. In that case, if we don't wake up the the network thread, it may not be able to send the next fetch request for a long time.

The enqueuing of other events will also call wakeupNetworkThread, so it shouldn't be a problem. But I went ahead and changed it as requested.

junrao · 2023-10-20T22:04:01Z

...rc/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java

+    /**
+     * Wakeup the {@link ConsumerNetworkThread network I/O thread} to pull the event from the queue.
+     */
+    public void wakeup() {


wakeup => wakeupNetworkThread ?

junrao · 2023-10-20T22:33:26Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetsRequestManager.java

+        Map<TopicPartition, Long> offsetResetTimestamps;
+
+        try {
+            offsetResetTimestamps = offsetFetcherUtils.getOffsetResetTimestamp();


auto.offset.reset has 3 options: latest, earliest, none. It's possible for a user to choose none, which means "throw exception to the consumer if no previous offset is found for the consumer's group". But in that case, offsetFetcherUtils.getOffsetResetTimestamp just ignores that partition and won't reset it forever. We need to throw an exception so that the user knows.

It's a bit of weird configuration since it means a consumer can't really consume for the very first time with this option. However, I tested it out with the existing consumer and it does throw an exception.

bin/kafka-console-consumer.sh --consumer-property auto.offset.reset=none --bootstrap-server localhost:9092 --topic test [2023-10-20 15:30:07,088] ERROR Error processing message, terminating consumer process: (kafka.tools.ConsoleConsumer$) org.apache.kafka.clients.consumer.NoOffsetForPartitionException: Undefined offset with no reset policy for partitions: [test-0] at org.apache.kafka.clients.consumer.internals.SubscriptionState.resetInitializingPositions(SubscriptionState.java:711) at org.apache.kafka.clients.consumer.KafkaConsumer.updateFetchPositions(KafkaConsumer.java:2459) at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1243) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1198) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1178) at kafka.tools.ConsoleConsumer$ConsumerWrapper.receive(ConsoleConsumer.scala:473) at kafka.tools.ConsoleConsumer$.process(ConsoleConsumer.scala:103) at kafka.tools.ConsoleConsumer$.run(ConsoleConsumer.scala:77) at kafka.tools.ConsoleConsumer$.main(ConsoleConsumer.scala:54) at kafka.tools.ConsoleConsumer.main(ConsoleConsumer.scala)

…OffsetForPartitionException if there's no reset strategy

kirktrue · 2023-10-20T22:53:03Z

FYI: I rebased against trunk to remove the conflicts.

junrao

@kirktrue : Thanks for the updated PR. Just a couple of minor comments.

junrao · 2023-10-20T22:56:48Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

+            timer.update(pollTimer.currentTimeMs());
+        }
+
+        return collectFetch();


Could we just return empty here instead of calling collectFetch() again since the caller is in a loop and can call this method to collect fetch again?

Yes, we could update pollForFetches() to simply return Fetch.empty() at the end.

However, there's code that intentionally blocks (for a little bit) waiting for fetch data to arrive in the buffer:

try { fetchBuffer.awaitNotEmpty(pollTimer); } catch (InterruptException e) { log.trace("Timeout during fetch", e); } finally { timer.update(pollTimer.currentTimeMs()); } return collectFetch();

Also, each loop through poll() executes updateAssignmentMetadataIfNeeded() before checking the fetch buffer for any data. That method does a lot of work (network I/O) that we'd ideally skip if we already have the data.

Perhaps we could update awaitNotEmpty() with a return flag so that the tail end of pollForFetches() can look something like this:

try { if (fetchBuffer.awaitNotEmpty(pollTimer)) return collectFetch(); } catch (InterruptException e) { log.trace("Timeout during fetch", e); } finally { timer.update(pollTimer.currentTimeMs()); } return Fetch.empty();

Thanks for the explanation, Kirk. We can just leave the code as it is then.

junrao · 2023-10-20T23:12:39Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

+        // positions, so a consumer with manually assigned partitions can avoid a coordinator
+        // dependence by always ensuring that assigned partitions have an initial position.
+        if (isCommittedOffsetsManagementEnabled() && !refreshCommittedOffsetsIfNeeded(timer))
+            return false;


Would it be clearer to rename refreshCommittedOffsetsIfNeeded to initWithCommittedOffsetsIfNeeded?

The refreshCommittedOffsetsIfNeeded() name is derived from existing code in KafkaConsumer that calls out to ConsumerCoordinator.refreshCommittedOffsetsIfNeeded(). If we change the method name in PrototypeAsyncConsumer, should we change ConsumerCoordinator's corresponding method name too?

Yes, in both cases, the method sets offsets for partitions in initializing state.

Updated and a bit of cleanup.

…Needed Also removed the check for null from ConsumerUtils.refreshCommittedOffsets() and put it on the ConsumerCoordinator.initWithCommittedOffsetsIfNeeded to check for null offsets map before calling.

junrao

@kirktrue : Thanks for the latest update. The PR LGTM now. Are the 30 test failures related?

kirktrue · 2023-10-23T22:16:56Z

Are the 30 test failures related?

No, I do not believe they are.

There are 8 "new" failures, of which two are the same test with different parameters. Here are the tests and any Jiras for flakiness:

kafka.server.DescribeClusterRequestTest.testDescribeClusterRequestExcludingClusterAuthorizedOperations: KAFKA-15419
kafka.server.DynamicBrokerReconfigurationTest.testLogCleanerConfig: KAFKA-7966, though the Jira is resolved 🤷‍♂️
o.a.k.common.network.SslVersionsTransportLayerTest.testTlsDefaults: KAFKA-9714, marked as "Critical" but open for 3 1/2 years
o.a.k.connect.integration.ConnectorRestartApiIntegrationTest.testMultiWorkerRestartOnlyConnector: KAFKA-15675, just filed (by me) with some pointers to a recent flakiness rate of 9%
o.a.k.streams.integration.StandbyTaskEOSMultiRebalanceIntegrationTest.shouldHonorEOSWhenUsingCachingAndStandbyReplicas: no Jira. Only around 2% flaky
o.a.k.tiered.storage.integration.TransactionsWithTieredStoreTest.testSendOffsetsWithGroupId: KAFKA-8003, though the Jira is filed against a different test, but test failures in testSendOffsetsWithGroupId are mentioned as being related
o.a.k.trogdor.coordinator.CoordinatorTest.testTaskRequestWithOldStartMsGetsUpdated: KAFKA-8115, which is another old, 4.5 year old "Critical" issue

There are 22 "existing" failures, which are all related to the verifyNoUnexpectedThreads check. kafka.server.DynamicBrokerReconfigurationTest is reporting that there are threads when it attempts to tear down the test harness. The remaining 21 failures all report the same unexpected threads in their check during test harness setup. There are other recent, unrelated pull requests that have experienced similar issues. In the other cases, I haven't seen that it's the DynamicBrokerReconfigurationTest that is the cause, though.

kirktrue · 2023-10-23T22:18:20Z

Closing and reopening to trigger another Jenkins test run.

kirktrue · 2023-10-24T02:05:31Z

Ugh. One of the builds failed with:

ERROR: Error fetching remote repo 'origin'
hudson.plugins.git.GitException: Failed to fetch from https://github.com/apache/kafka.git
	at hudson.plugins.git.GitSCM.fetchFrom(GitSCM.java:1003)
	at hudson.plugins.git.GitSCM.retrieveChanges(GitSCM.java:1245)
	at hudson.plugins.git.GitSCM.checkout(GitSCM.java:1309)
	at org.jenkinsci.plugins.workflow.steps.scm.SCMStep.checkout(SCMStep.java:129)
	at org.jenkinsci.plugins.workflow.steps.scm.SCMStep$StepExecutionImpl.run(SCMStep.java:97)
	at org.jenkinsci.plugins.workflow.steps.scm.SCMStep$StepExecutionImpl.run(SCMStep.java:84)
	at org.jenkinsci.plugins.workflow.steps.SynchronousNonBlockingStepExecution.lambda$start$0(SynchronousNonBlockingStepExecution.java:47)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)

No integration test failures due to threads, though 😏

kirktrue · 2023-10-24T02:05:48Z

Closing and reopening to restart test.

kirktrue · 2023-10-24T15:13:43Z

Closing and reopening to restart test. Jenkins doesn't seem happy lately.

lianetm · 2023-10-24T15:20:23Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java

    @Override
    public List<PartitionInfo> partitionsFor(String topic) {
-        throw new KafkaException("method not implemented");
+        return partitionsFor(topic, Duration.ofMillis(defaultApiTimeoutMs));


I had missed a detail here. We're only implementing this with the default timeout, but not the overloaded one below, that has the timeout provided by the user. Is there a reason or is it just that we missed the latter one? (same for the listTopics)

(I guess it will be all part of the integration with the metadata calls right?)

Good call out. I don't know that the necessary topic plumbing code was written at that time. Would you mind filing a bug to resolve?

junrao · 2023-10-24T19:04:42Z

@kirktrue : The build for JDK 21 and Scala 2.13 failed this time. Did it succeed before?

kirktrue · 2023-10-24T19:28:41Z

@junrao—yes. Here's a brief history for JDK 21, starting with the most recent build (build 74):

build 74: couldn't clone repo
build 73: tests ran, but Jenkins timed out
build 72: tests ran
build 71: tests ran
build 70: tests ran, but Jenkins shut down in the middle 🤔

These same intra-Jenkins communication, git cloning, unexpected threads, and flaky tests affect all of the JDKs at random times 😢

junrao

@kirktrue : Thanks for explaining the test failures. The PR LGTM.

Changes: 1. Introduces FetchRequestManager that implements the RequestManager API for fetching messages from brokers. Unlike Fetcher, record decompression and deserialization is performed on the application thread inside CompletedFetch. 2. Restructured the code so that objects owned by the background thread are not instantiated until the background thread runs (via Supplier) to ensure that there are no references available to the application thread. 3. Ensuring resources are properly using Closeable and using IdempotentCloser to ensure they're only closed once. 4. Introduces ConsumerTestBuilder to reduce a lot of inconsistency in the way the objects were built up for tests. Reviewers: Philip Nee <pnee@confluent.io>, Lianet Magrans <lianetmr@gmail.com>, Jun Rao<junrao@gmail.com>

kirktrue added 2 commits September 19, 2023 10:24

Reverted change to make FetchCollector.collectFetch() not take a Dese…

7ab56f2

…rializers

philipnee added the ctr Consumer Threading Refactor (KIP-848) label Sep 20, 2023

philipnee reviewed Sep 21, 2023

View reviewed changes

checkstyle/import-control.xml Show resolved Hide resolved

philipnee reviewed Sep 21, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ErrorEventHandler.java Outdated Show resolved Hide resolved

philipnee reviewed Sep 21, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchRequestManager.java Outdated Show resolved Hide resolved

philipnee reviewed Sep 21, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java Outdated Show resolved Hide resolved

philipnee reviewed Sep 21, 2023

View reviewed changes

...c/main/java/org/apache/kafka/clients/consumer/internals/events/BackgroundEventProcessor.java Outdated Show resolved Hide resolved

lianetm reviewed Sep 21, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java Outdated Show resolved Hide resolved

lianetm reviewed Sep 21, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java Outdated Show resolved Hide resolved

lianetm reviewed Sep 21, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java Outdated Show resolved Hide resolved

philipnee reviewed Sep 21, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java Outdated Show resolved Hide resolved

lianetm reviewed Sep 21, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultBackgroundThread.java Outdated Show resolved Hide resolved

lianetm reviewed Sep 21, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/DefaultEventHandler.java Outdated Show resolved Hide resolved

junrao reviewed Sep 22, 2023

View reviewed changes

kirktrue added 13 commits September 22, 2023 15:00

Reverted change to make FetchCollector.collectFetch() not take a Dese…

a30ac4e

…rializers

Post-rebase clean up

72b6b27

Merge branch 'KAFKA-14274-3-introduce-fetch-request-manager-take-2' o…

ba9f0ff

…f github.com:kirktrue/kafka into KAFKA-14274-3-introduce-fetch-request-manager-take-2

Update NetworkClientDelegate.java

7c8b945

Minor clean up and refactoring based on first pass of reviews

59d0daf

Merge branch 'trunk' into KAFKA-14274-3-introduce-fetch-request-manag…

0457117

…er-take-2

Changes to PrototypeAsyncConsumer.close()

1e18f98

Refactored the close method so that it's much cleaner. Also fixed a confusing doc change to refreshCommittedOffsetsIfNeeded().

Minor refactoring for the event processor callbacks

e8b990f

Added comment to explain a case in DefaultBackgroundThread.wakeup()

731e716

philipnee reviewed Sep 27, 2023

View reviewed changes

clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java Show resolved Hide resolved

kirktrue added 2 commits October 20, 2023 14:53

Making NetworkClientDelegate constructor call another to eliminate re…

f40d4d2

…dundancy

Merge branch 'trunk' into KAFKA-14274-3-introduce-fetch-request-manag…

56d0b40

…er-take-2

junrao reviewed Oct 20, 2023

View reviewed changes

kirktrue added 2 commits October 20, 2023 15:45

Ensure wakeupNetworkThread() is called even on empty Fetch results

13cb3b6

Changed OffsetFetcherUtils.offsetResetStrategyTimestamp to throw a No…

9525fca

…OffsetForPartitionException if there's no reset strategy

junrao reviewed Oct 20, 2023

View reviewed changes

Renamed refreshCommittedOffsetsIfNeeded to initWithCommittedOffsetsIf…

c5d5fc6

…Needed Also removed the check for null from ConsumerUtils.refreshCommittedOffsets() and put it on the ConsumerCoordinator.initWithCommittedOffsetsIfNeeded to check for null offsets map before calling.

junrao reviewed Oct 21, 2023

View reviewed changes

kirktrue closed this Oct 23, 2023

kirktrue reopened this Oct 23, 2023

kirktrue closed this Oct 24, 2023

kirktrue reopened this Oct 24, 2023

kirktrue closed this Oct 24, 2023

kirktrue reopened this Oct 24, 2023

lianetm reviewed Oct 24, 2023

View reviewed changes

junrao approved these changes Oct 24, 2023

View reviewed changes

junrao merged commit 2b233bf into apache:trunk Oct 24, 2023

kirktrue deleted the KAFKA-14274-3-introduce-fetch-request-manager-take-2 branch October 25, 2023 00:14

This was referenced Dec 14, 2023

KAFKA-15556: Remove NetworkClientDelegate methods isUnavailable, maybeThrowAuthFailure, and tryConnect #14979

Closed

KAFKA-15556: Remove NetworkClientDelegate methods isUnavailable, maybeThrowAuthFailure, and tryConnect #15020

Closed

Conversation

kirktrue commented Sep 19, 2023

Uh oh!

kirktrue commented Sep 20, 2023

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kirktrue commented Oct 20, 2023

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

kirktrue commented Oct 23, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kirktrue commented Oct 23, 2023

Uh oh!

kirktrue commented Oct 24, 2023

Uh oh!

kirktrue commented Oct 24, 2023

Uh oh!

kirktrue commented Oct 24, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kirktrue commented Oct 23, 2023 •

edited

Loading