KAFKA-15181: Improvements for TopicBaseRemoteLogMetadataManager by abhijeetk88 · Pull Request #14127 · apache/kafka

abhijeetk88 · 2023-07-29T05:48:50Z

This PR adds the following changes to the TopicBasedRemoteLogMetadataManager

Added a guard in RemoteLogMetadataCache so that the incoming request can be served from the cache iff the corresponding user-topic-partition is initalized
Improve error handling in ConsumerTask thread so that is not killed when there are errors in reading the internal topic
ConsumerTask initialization should handle the case when there are no records to read

and some other minor changes

Added Unit Tests for the changes

Co-authored-by: Kamal Chandraprakash kamal.chandraprakash@gmail.com

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

abhijeetk88 · 2023-08-02T06:32:16Z

@junrao @showuon @jeqo @kamalcph can you help in reviewing this PR?

showuon · 2023-08-02T07:01:20Z

Will check later this week

showuon

Will continue review later. Thanks.

showuon · 2023-08-04T03:10:27Z

+    }
+
+    public void ensureInitialized() throws InterruptedException {
+        if (!initializedLatch.await(2, TimeUnit.MINUTES)) {


I can see we will block for operations like findOffsetByTimestamp (invoked by clients), and the client request timeout default is 30 secs. I think the 2 mins wait is not appropriate here. Maybe 10 secs? Or any reason we chose 2 mins?

In large clusters, we saw that it takes more than a minute to initialize. That's the reason we kept it 2 mins. We could also make this configurable. Thoughts?
cc @kamalcph

Both FETCH and LIST_OFFSETS APIs will hit this method. If the partition is not initialised within the timeout (2 mins / 10 secs), then we throw RemoteResourceNotFoundException which will in turn throw UNKNOWN_SERVER_ERROR back to the caller.

Consumer will retry the FETCH request as UNKNOWN_SERVER_ERROR is retriable. The LIST_OFFSETS call can originate from both Consumer and Admin client. Consumer retries the request on receiving retriable error but Admin client won't. If the user tries to increase the request timeout from 30 secs to 2 min in AdminClient, then the LIST_OFFSETS request should succeed.

This timeout specified here is kind of upper boundary where we expect the RLMM to initialise the partition within that amount of time.

If the request timeout is configured to default value of 30 secs on the client, then the request gets timed out and the client retries the request. In case of AdminClient, the user gets to know that the request was failed due to TimeoutException and will increase the request timeout as appropriate.

We can reduce the timeout to 10 seconds but should think of a solution for Admin#listOffsets call.

Thanks for the explanation, @kamalcph !

One correction: UNKNOWN_SERVER_ERROR is unretriable. So if we throw RemoteResourceNotFoundException, the client won't be retried.

I think we don't have to consider if client will auto retry or not when facing retriable error. IMO, sending a request and returned a request timeout error without any explanation (suppose the 2 min timeout is not expired here, so no error thrown in the broker side), which is not good.

I think reduce to a small number (I think 10 secs is still too long, maybe 1 sec, or return error immediately if it's not initialized), and wrap the RemoteResourceNotFoundException with a retriable error, ex: ReplicaNotAvailableException, and add clear messages there. I think as long as we returned clear error reason with a retriable exception, this is already a good API. We don't want to block the request for a long time because we're waiting for some state change. Like COORDINATOR_LOAD_IN_PROGRESS error, it's clear that we returned this error and want the client retry later, instead of waiting in the broker for coordinator load completion. Does that make sense?

The java client retries when it receives UNKNOWN_SERVER_ERROR not sure for other clients. As per the protocol contract, the clients are not expected to retry when it receives this error.

We also have one more API isInitialized that can be used instead of ensureInitialized if we want to unblock the caller immediately. This latch will only block the remote-log-reader and remote-log-deleter threads until the partition gets initialised and won't block the request handler threads.

The idea to increase the timeout upto 2 minutes was similar to delayed purgatory where the requests are hold until it gets served within the request timeout.

Currently, the LIST_OFFSETS call is sync which is handled by the request handler threads so the request-handler-thread will be blocked here. Until KAFKA-13560 gets implemented, it's good to reduce the timeout to 1s (or) throw error immediately as suggested. Also, it will unblock the remote-log-reader threads where it can be utilised for other initialized partitions.

Also, ListOffsetsHandler handles retriable error. So, we are good if we throw retryable error back to the client.

@kamalcph , thanks for the update!
@abhijeetk88 , could you reduce the timeout to 1s (or) throw error immediately ( I think use isInitialized() direcrtly is better) as discussed above? Thanks.

@showuon I have addressed this. Added a new retryable exception. Let me know if we need to rename it.

showuon · 2023-08-04T03:13:40Z

+        // If the leader for a `__remote_log_metadata` partition is not available, then the call to `ListOffsets`
+        // will fail after the default timeout of 1 min. Added a delay of 5 min in between the retries to prevent the
+        // thread from aggressively fetching the list offsets. During this time, the recently reassigned
+        // user-topic-partitions won't be marked as initialized.
+        if (isOffsetsFetchFailed && lastFailedFetchOffsetsTimestamp + 300_000 < time.milliseconds()) {
+            fetchBeginAndEndOffsets();
+        }


The 5 mins wait is too long. Like you said in the comment, timeout is 1 min, why can't we retry after 10 secs or so to have more chance to get the offsets? IMO, 10 secs for electing a leader for a partition is long enough, so we should reduce this timeout. Thought?

When the leader is offline, it may take some time to come up. In that case, we want to avoid aggressively fetching the offsets. Otherwise, we will frequently call this method and get blocked for 1 min (until the API times out). This will prevent us from processing updates for the already assigned topic partitions.

It is basically a tradeoff (when the leader is offline) between whether we want to update for the newly assigned topic partitions as soon as possible or always be up-to-date for the already assigned topic partitions. We chose the second option here.

Could we make this a ConsumerTask input parameter? Maybe later can be exposed as a configuration.

Added input parameter

showuon

@abhijeetk88 , thanks for the PR. Left some comments.

showuon · 2023-08-04T06:13:15Z

+    @Test
+    public void testCloseOnNoAssignment() throws InterruptedException {
+        thread.start();
+        Thread.sleep(10);


Could you explain what this test is trying to verify? If we want to verify the consumer thread is closed, should we explicitly assert that?

It is trying to verify that when there were no assignments made, the consumer thread still shuts down gracefully.

Added comments

Could we add some assertion to verify consumer task shuts down gracefully as you expected? In current test, even if the consumer task doesn't shutdown, it sill passes the test, right?

I will add assertions in 'afterEach' method to ensure that the consumer thread shuts down gracefully

It's still unclear in this test, because the assertion you added is not inside the test. Could you at least add an assertion inside the test, ex:

@Test public void testCloseOnNoAssignment() throws InterruptedException { thread.start(); Thread.sleep(10); assertDoesNotThrow(() -> consumerTask.close(), "Close method threw exception");

This way, people can understand you want to verify it should be closed gracefully.

Added assertion for clarity.

abhijeetk88 · 2023-08-08T06:08:47Z

Thanks for the review @showuon. I have addressed your comments and responded to your questions. Please take a look.

clolov · 2023-08-08T13:22:07Z

Apologies for joining the review late (I had to spend a few hours getting acquainted with the ConsumerTask code). I carried out a first pass on the changes to the main files and they make sense to me. I will circle back tomorrow morning and have another go through the tests.

clolov

Okay, I finished my first pass through the tests and I have a couple of small comments/questions

clolov · 2023-08-09T13:51:36Z

+        final List<TopicIdPartition> idPartitions = getIdPartitions("sample", 3);
+        final Map<TopicPartition, Long> endOffsets = idPartitions.stream()
+            .map(idp -> toRemoteLogPartition(partitioner.metadataPartition(idp)))
+            .collect(Collectors.toMap(Function.identity(), e -> 0L, (a, b) -> b));


Is the merge function needed? If not, can we get rid of it, because its presence leads me to think that we can have multiple values associated with the same key and unless I am wrong that isn't the case here?

The TopicIdPartitions are being converted to metadata topic partitions and this will result in multiple user topic partitions getting mapped to the same metadata topic partition. The merge function is required to handle multiple entries with the same key. Here we choose any value in the merge function, since all of the values will be 0.

clolov · 2023-08-09T14:36:50Z

+            0, 100, -1L, 0,
+            time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 0L));
+        Assertions.assertThrows(ExecutionException.class, () -> rlmm().addRemoteLogSegmentMetadata(leaderSegmentMetadata).get(),
+            "org.apache.kafka.common.KafkaException: This consumer is not assigned to the target partition 0. Partitions currently assigned: []");


I thought that the last argument to an assertThrows is the message you would like to return if the assertion fails rather than the message you would like to check is contained in the exception? Or am I misunderstanding the purpose of this message?

Can you give a bit more explanation here, because I do not understand this? Are we expecting this to throw, in which case, why would it accept the metadata entry or are we not expecting it to throw?

My bad, I actually want to assert that it fails with that message. I will change this.

We are trying to assert that adding the remote log segment metadata entry will throw an exception since the partition is not subscribed yet.

jeqo

Thanks @abhijeetk88. Took an initial look and left some comments. Most changes look good to me. Will have another look at ConsumerTask (without looking at git diff 😅 ) and tests tomorrow.

jeqo · 2023-08-09T14:38:40Z

    private static final Logger log = LoggerFactory.getLogger(ConsumerTask.class);

-    private static final long POLL_INTERVAL_MS = 100L;
+    static long pollIntervalMs = 100L;


Is this really needed?
I can see value is overwritten on test setup to 10L, but I can manage to test successfully with default value 100L.

It will take longer for the tests to run with higher poll intervals. That is the reason we want to override it in tests.

Got it. Though, I'd argue a mutable static field may not be the best option here. Adding pollIntervalMs as input parameter/config may be better?

Also, isn't 100ms too often to poll again? Maybe a few seconds or higher may be a better default.

+1 on pollIntervalMs should not be a static field here. We can also have this as a config if needed later.

Added as an input parameter.

Also, it is a timeout value and not a poll interval. I have updated the field name. Let me know if the default value of 100 ms does not make sense for timeout.

I'd suggest increasing the timeout to at least 5 seconds to reduce poll frequency when no new data is available.

jeqo · 2023-08-09T14:44:06Z

-                        RemotePartitionMetadataEventHandler remotePartitionMetadataEventHandler,
+    public ConsumerTask(RemotePartitionMetadataEventHandler remotePartitionMetadataEventHandler,
                        RemoteLogMetadataTopicPartitioner topicPartitioner,
-                        Path committedOffsetsPath,


Haven't look to deep into this, but could you elaborate on why this file is not needed anymore?

btw, COMMITTED_OFFSETS_FILE_NAME can be removed on ConsumerManager.

We want to keep the implementation simple for now without checkpointing mechanism.

abhijeetk88 · 2023-08-12T10:20:50Z

@jeqo @showuon I have addressed your comments. Please take a look.

jeqo · 2023-08-14T21:48:53Z

+        }
+    }
+
+    static class StartAndEndOffsetHolder {


nit: not sure the Holder adds much, but feel free to ignore this.

Suggested change

static class StartAndEndOffsetHolder {

static class StartAndEndOffsets {

jeqo · 2023-08-14T21:58:41Z

+        // If the leader for a `__remote_log_metadata` partition is not available, then the call to `ListOffsets`
+        // will fail after the default timeout of 1 min. Added a delay of 5 min in between the retries to prevent the
+        // thread from aggressively fetching the list offsets. During this time, the recently reassigned
+        // user-topic-partitions won't be marked as initialized.
+        if (isOffsetsFetchFailed && lastFailedFetchOffsetsTimestamp + 300_000 < time.milliseconds()) {
+            fetchBeginAndEndOffsets();
+        }


Could we make this a ConsumerTask input parameter? Maybe later can be exposed as a configuration.

jeqo · 2023-08-14T22:38:21Z

    private static final Logger log = LoggerFactory.getLogger(ConsumerTask.class);

-    private static final long POLL_INTERVAL_MS = 100L;
+    static long pollIntervalMs = 100L;


Got it. Though, I'd argue a mutable static field may not be the best option here. Adding pollIntervalMs as input parameter/config may be better?

Also, isn't 100ms too often to poll again? Maybe a few seconds or higher may be a better default.

jeqo

Thanks @abhijeetk88. Adding a few more comments while taking another round of reviews on the tests.

satishd

Thanks @abhijeetk88 for the PR, left a minor comment.

satishd · 2023-08-20T05:16:50Z

    private static final Logger log = LoggerFactory.getLogger(ConsumerTask.class);

-    private static final long POLL_INTERVAL_MS = 100L;
+    static long pollIntervalMs = 100L;


+1 on pollIntervalMs should not be a static field here. We can also have this as a config if needed later.

…gning partitions

abhijeetk88 · 2023-08-25T03:45:49Z

Thanks @showuon for your comments. I have addressed them inline/with the latest commits. Please take a look.

Also, rebased the changes on top of the latest tunk.

showuon

Overall LGTM! Just one comment. Thanks.

showuon · 2023-08-25T07:38:41Z

@@ -0,0 +1,14 @@
+package org.apache.kafka.common.errors;
+
+public class ResourceNotReadyException extends RetriableException {


Sorry, org.apache.kafka.common.errors is a public API, we need a KIP to add it. That's why I suggested to use the existing ReplicaNotAvailableException instead. Could we change it and we can write a KIP for this new ResourceNotReadyException if you think it's more appropriate?

+1 on not introducing a new error as it requires a KIP.
We can reuse the closest possible error for now. ReplicaNotAvailableException triggers refreshing a metadata update from the client as it is kind of an InvalidMetadataException which may generate a lot of metadata calls. It may be good to return an exception that does not call for metadata refresh.

We can introduce a new error as a separate KIP later and handle it appropriately.

We couldn't find another exception that fits our case. We can use ReplicaNotAvailableException for now. I have created a jira to introduce a new retryable exception/error code.
https://issues.apache.org/jira/browse/KAFKA-15405

That looks reasonable to me for now.

Updated the PR.

showuon

LGTM! Just a minor comment. Thanks for the patch!

satishd

Thanks @abhijeetk88 for addressing the review comments. LGTM.

satishd · 2023-08-26T00:19:25Z

It seems test failures are unrelated to this PR. Merging it to trunk.

satishd · 2023-08-26T00:23:39Z

Merging to 3.6 also.

junrao · 2024-01-10T00:15:15Z

+    private final Map<TopicIdPartition, Long> readOffsetsByUserTopicPartition = new HashMap<>();

-    private final long committedOffsetSyncIntervalMs;
-    private CommittedOffsetsFile committedOffsetsFile;


@abhijeetk88 @satishd : Sorry to chime in late on this. It seems that we removed the usage of CommittedOffsetsFile.

Does the consumer always start from the beginning offset after restart?

Should we remove the CommittedOffsetsFile class?

This actually changes the on-disk layout. Was this discussed in a KIP?

@junrao , answering your question:

Does the consumer always start from the beginning offset after restart?
Yes, we always read from beginning each time we restart.

Should we remove the CommittedOffsetsFile class?
Good point. Will remove it in another PR.

This actually changes the on-disk layout. Was this discussed in a KIP?

I had a look, and yes, the committed offset file is mentioned in the KIP-405. @kamalcph @satishd , do we remove the offset committed file and read from beginning each time on purpose? To fix the original bug, I don't think we need to remove this offset committed file mechanism. WDYT?

Sorry for the late reply. Now, we are always reading from the beginning of the topic to restore the state. And, the committed offsets design is unused and kept as-is for future improvement. Storing the snapshot of the remote log metadata under each partition requires some discussion.

How often to trigger the snapshot?

If we configure to run the snapshots in a scheduled interval, then it can lead to flushing of the log segments which is async now.

On broker shutdown: This can lead to two cases: clean and unclean shutdown. During unclean shutdown, we cannot depend on the snapshot stored in the local disk as it can be corrupted. Reading from the source-of-truth (__remote_log_metadata topic) is reliable (v0).

On Partition reassignments/New-Topic: When a user-topic-partition (utp) is reassigned to another broker, it has to reload the context from the __remote_log_metadata topic.

For V0, we kept the design to read from the beginning of the respective metadata topic-partition.

Sorry @junrao for missing this comment in my list of notifications.

Adding to what @showuon and @kamalcph mentioned above.

Should we remove the CommittedOffsetsFile class?

We may want to keep it in the repo and use it in the future or we can remove it for now and add it back when we work on this in future. I am fine with either way.

@kamalcph @satishd @showuon : Thanks for the reply. Do we plan to use CommittedOffsetsFile anytime soon? If not, it's probably better to remove it for now and add it back when it's being used. Also, have we decided if CommittedOffsetsFile will be used in the same way as described in KIP-405? If not, it would be useful to update the KIP with the new design and send the update to the vote thread.

@junrao We will remove the CommittedOffsetsFile for now as you suggested. The current plan is to use it as mentioned in KIP-405. We will follow KIP update process and notify in the vote thread if it is different from the existing proposal.

abhijeetk88 marked this pull request as draft July 29, 2023 05:48

abhijeetk88 force-pushed the rlmm_improvements branch from 67f35bb to 1a86b92 Compare July 29, 2023 07:00

abhijeetk88 changed the title ~~KAFKA-15181: Wait for RemoteLogMetadataCahce to initialize after assi…~~ KAFKA-15181: Wait for RemoteLogMetadataCache to initialize after assi… Jul 29, 2023

abhijeetk88 changed the title ~~KAFKA-15181: Wait for RemoteLogMetadataCache to initialize after assi…~~ KAFKA-15181: Wait for RemoteLogMetadataCache to initialize after assigning partitions Jul 30, 2023

This was referenced Jul 29, 2023

KAFKA-15261: Do not block replica fetcher if RLMM is not initialized #14112

Closed

KAFKA-15260: RLM Task should handle uninitialized RLMM for the associated topic-parititon #14113

Merged

clolov added the tiered-storage Related to the Tiered Storage feature label Jul 31, 2023

abhijeetk88 marked this pull request as ready for review August 1, 2023 05:40

showuon reviewed Aug 3, 2023

View reviewed changes

Comment thread storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java Outdated

Comment thread storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java Outdated

showuon reviewed Aug 3, 2023

View reviewed changes

Comment thread storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java Outdated

Comment thread storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java Outdated

showuon reviewed Aug 4, 2023

View reviewed changes

abhijeetk88 force-pushed the rlmm_improvements branch from 812f7c4 to 4121b0d Compare August 8, 2023 06:02

mdedetrich mentioned this pull request Aug 8, 2023

Race condition between ReplicaFetcherThread and RemotePartitionMetadataStore. aiven/kafka#33

Closed

clolov reviewed Aug 9, 2023

View reviewed changes

jeqo reviewed Aug 9, 2023

View reviewed changes

abhijeetk88 force-pushed the rlmm_improvements branch 2 times, most recently from 44a794f to bcc6f56 Compare August 12, 2023 10:13

jeqo reviewed Aug 14, 2023

View reviewed changes

This was referenced Aug 15, 2023

KIP-405: 2023-08-15 aiven/kafka#36

Closed

KIP-405: 2023-08-18 aiven/kafka#37

Closed

satishd reviewed Aug 20, 2023

View reviewed changes

abhijeetk88 force-pushed the rlmm_improvements branch from eff6247 to fa93a24 Compare August 20, 2023 12:18

abhijeetk88 added 7 commits August 25, 2023 11:28

KAFKA-15181: Wait for RemoteLogMetadataCache to initialize after assi…

619273f

…gning partitions

Address review comments

403c355

Address review comments

095cfae

Address review comments

692895c

Review comments

83c7e74

Added assert statement for clarity

00f5697

Introduce a retryable error

6b84359

abhijeetk88 force-pushed the rlmm_improvements branch from da7aa6a to 6b84359 Compare August 25, 2023 03:43

showuon reviewed Aug 25, 2023

View reviewed changes

Use ReplicaNotAvailableException instead for time-being

3639651

showuon approved these changes Aug 25, 2023

View reviewed changes

Comment thread ...n/java/org/apache/kafka/server/log/remote/metadata/storage/RemotePartitionMetadataStore.java

Add comment for introducing a new retryable exception later

9391ac7

satishd approved these changes Aug 25, 2023

View reviewed changes

satishd merged commit ff3e684 into apache:trunk Aug 26, 2023

kamalcph mentioned this pull request Aug 28, 2023

MINOR: Fix the TBRLMMRestart test. #14297

Merged

3 tasks

abhijeetk88 changed the title ~~KAFKA-15181: Wait for RemoteLogMetadataCache to initialize after assigning partitions~~ KAFKA-15181: Improvement for TopicBaseRemoteLogMetadataManager Sep 5, 2023

abhijeetk88 changed the title ~~KAFKA-15181: Improvement for TopicBaseRemoteLogMetadataManager~~ KAFKA-15181: Improvements for TopicBaseRemoteLogMetadataManager Sep 5, 2023

mdedetrich mentioned this pull request Sep 11, 2023

KAFKA-14975: Wait for TopicBasedRemoteLogMetadataManager.ensureInitializedAndNotClosed to initialize rather than throwing #13689

Closed

3 tasks

junrao reviewed Jan 10, 2024

View reviewed changes

	static class StartAndEndOffsetHolder {
	static class StartAndEndOffsets {

		@@ -0,0 +1,14 @@
		package org.apache.kafka.common.errors;

		public class ResourceNotReadyException extends RetriableException {

Uh oh!

Conversation

abhijeetk88 commented Jul 29, 2023 • edited by satishd Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

abhijeetk88 commented Aug 2, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

showuon commented Aug 2, 2023

Uh oh!

showuon 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!

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!

showuon Aug 21, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kamalcph Aug 21, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

showuon Aug 22, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

showuon left a comment

Choose a reason for hiding this comment

Uh oh!

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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

abhijeetk88 commented Jul 29, 2023 •

edited by satishd

Loading

abhijeetk88 commented Aug 2, 2023 •

edited

Loading

showuon Aug 21, 2023 •

edited

Loading

kamalcph Aug 21, 2023 •

edited

Loading

showuon Aug 22, 2023 •

edited

Loading

abhijeetk88 Aug 12, 2023 •

edited

Loading

abhijeetk88 Aug 11, 2023 •

edited

Loading

jeqo left a comment •

edited

Loading

satishd Aug 20, 2023 •

edited

Loading