Skip to content

KAFKA-16557 Implemented OffsetFetchRequestState toStringBase and added a test for it #16291

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chia7712 merged 11 commits into from
Jun 13, 2024
Merged

KAFKA-16557 Implemented OffsetFetchRequestState toStringBase and added a test for it #16291

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
81a5572
Add toStringBase(), toString(), and a test
brenden20 Jun 11, 2024
7147a27
Cleaner line
brenden20 Jun 11, 2024
e91891f
Add toStringBase() to RetriableRequestState
brenden20 Jun 11, 2024
caf17bb
Remove print statement
brenden20 Jun 11, 2024
1941a55
Small update
brenden20 Jun 11, 2024
5768017
Removed unnecessary setter method
brenden20 Jun 11, 2024
94f26b5
Merge branch 'apache:trunk' into 16557-2
brenden20 Jun 12, 2024
837d9ea
Add assertFalse statement
brenden20 Jun 12, 2024
98361ea
Fixed toStringBase() test
brenden20 Jun 12, 2024
f061d97
Merge branch '16557-2' of https://github.com/brenden20/kafka into 165…
brenden20 Jun 12, 2024
2583e12
Changed method visibility
brenden20 Jun 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 14 additions & 8 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -865,6 +865,11 @@ private void handleClientResponse(final ClientResponse response,
}
}

@Override
public String toStringBase() {
return super.toStringBase() + ", " + memberInfo;
}

abstract void onResponse(final ClientResponse response);

abstract void removeRequest();
Expand Down Expand Up @@ -1078,14 +1083,9 @@ private void chainFuture(
}

@Override
public String toString() {
return "OffsetFetchRequestState{" +
"requestedPartitions=" + requestedPartitions +
", memberId=" + memberInfo.memberId.orElse("undefined") +
", memberEpoch=" + (memberInfo.memberEpoch.isPresent() ? memberInfo.memberEpoch.get() : "undefined") +
", future=" + future +
", " + toStringBase() +
'}';
public String toStringBase() {
return super.toStringBase() +
", requestedPartitions=" + requestedPartitions;
}
}

Expand Down Expand Up @@ -1278,5 +1278,11 @@ static class MemberInfo {
this.memberId = Optional.empty();
this.memberEpoch = Optional.empty();
}

@Override
public String toString() {
return "memberId=" + memberId.orElse("undefined") +
", memberEpoch=" + (memberEpoch.isPresent() ? memberEpoch.get() : "undefined");
}
}
}
43 changes: 43 additions & 0 deletions ...s/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,7 @@
import static org.apache.kafka.clients.consumer.internals.ConsumerTestBuilder.DEFAULT_GROUP_ID;
import static org.apache.kafka.clients.consumer.internals.ConsumerTestBuilder.DEFAULT_GROUP_INSTANCE_ID;
import static org.apache.kafka.test.TestUtils.assertFutureThrows;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
Expand Down Expand Up @@ -122,6 +123,48 @@ public void setup() {
this.props.put(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
}

@Test
public void testOffsetFetchRequestStateToStringBase() {
ConsumerConfig config = mock(ConsumerConfig.class);
CommitRequestManager.MemberInfo memberInfo = new CommitRequestManager.MemberInfo();

@chia7712 chia7712 Jun 11, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

As we had discussion about "Optional#toString", maybe we define either MemberInfo#memberId or MemberInfo#memberEpoch, and then we should check the MemberInfo#toString does not contain "Optional[value]"

@brenden20 brenden20 Jun 11, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I see what you meant now, I have implemented a setter for MemberInfo#memberEpoch and set the memberEpoch value. Upon testing, there is no "Optional[value]"

@lianetm lianetm Jun 11, 2024 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

@brenden20 we already have an onMemberEpochUpdated that you could maybe use for this? so we don't have to add a new setter. That's the actual method that gets called when the member gets a new epoch on the heartbeat response, and we want to pass it on to the commit manager ;)

@brenden20 brenden20 Jun 11, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@lianetm oh I see it now, I will revert that change and use onMemberEpochUpdated instead. Thank you!


CommitRequestManager commitRequestManager = new CommitRequestManager(
time,
logContext,
subscriptionState,
config,
coordinatorRequestManager,
offsetCommitCallbackInvoker,
"groupId",
Optional.of("groupInstanceId"),
metrics);

commitRequestManager.onMemberEpochUpdated(Optional.of(1), Optional.empty());
Set<TopicPartition> requestedPartitions = Collections.singleton(new TopicPartition("topic-1", 1));

CommitRequestManager.OffsetFetchRequestState offsetFetchRequestState = commitRequestManager.new OffsetFetchRequestState(
requestedPartitions,
retryBackoffMs,
retryBackoffMaxMs,
1000,
memberInfo);

@chia7712 chia7712 Jun 11, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

in the line#142 - The memberInfo updated by onMemberEpochUpdated is the inner variable of commitRequestManager. Hence, this memberInfo used to create OffsetFetchRequestState is NOT updated by onMemberEpochUpdated

If we want to complete the test, maybe we can remove MemberInfo from constructor of OffsetFetchRequestState. After all, both RetriableRequestState and OffsetFetchRequestState are the inner class of CommitRequestManager, so it is valid to share the MemberInfo between those classes.

@brenden20 brenden20 Jun 12, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I am a bit hesitant to change the constructor of OffsetFetchRequestState as I do not want to mess up usages elsewhere in the codebase. I did however find a way to achieve updating the inner memberInfo. Test is working as well, I just needed to change the method from private to protected. Let me know what you think


TimedRequestState timedRequestState = new TimedRequestState(
logContext,
"CommitRequestManager",
retryBackoffMs,
retryBackoffMaxMs,
TimedRequestState.deadlineTimer(time, 0)
);

String target = timedRequestState.toStringBase() +
", " + memberInfo +
", requestedPartitions=" + offsetFetchRequestState.requestedPartitions;

assertDoesNotThrow(timedRequestState::toString);
assertEquals(target, offsetFetchRequestState.toStringBase());
}

@chia7712 chia7712 Jun 11, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Maybe we can add assertFalse(target.contains("Optional")); to make sure we unwrap the optional variables?

@brenden20 brenden20 Jun 12, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Added!


@Test
public void testPollSkipIfCoordinatorUnknown() {
CommitRequestManager commitRequestManager = create(false, 0);
Expand Down