You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "johnnychhsu (via GitHub)" <gi...@apache.org> on 2024/05/15 15:10:32 UTC
[PR] KAFKA-16684: Remove cache in responseData [kafka]
johnnychhsu opened a new pull request, #15966:
URL: https://github.com/apache/kafka/pull/15966
## Context
The response data should change accordingly to the input, however with the current design, it will not change even if the input changes. We should remove this cache logic to avoid returning wrong data.
Jira: https://issues.apache.org/jira/browse/KAFKA-16684
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
Re: [PR] KAFKA-16684: Remove cache in responseData [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #15966:
URL: https://github.com/apache/kafka/pull/15966#issuecomment-2206969288
```
org.gradle.api.internal.tasks.testing.TestSuiteExecutionException: Could not complete execution for Gradle Test Executor 100.
at org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.stop(SuiteTestClassProcessor.java:64)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36)
at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
at org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33)
at org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94)
at com.sun.proxy.$Proxy2.stop(Unknown Source)
at org.gradle.api.internal.tasks.testing.worker.TestWorker$3.run(TestWorker.java:193)
at org.gradle.api.internal.tasks.testing.worker.TestWorker.executeAndMaintainThreadName(TestWorker.java:129)
at org.gradle.api.internal.tasks.testing.worker.TestWorker.execute(TestWorker.java:100)
at org.gradle.api.internal.tasks.testing.worker.TestWorker.execute(TestWorker.java:60)
at org.gradle.process.internal.worker.child.ActionExecutionWorker.execute(ActionExecutionWorker.java:56)
at org.gradle.process.internal.worker.child.SystemApplicationClassLoaderWorker.call(SystemApplicationClassLoaderWorker.java:119)
at org.gradle.process.internal.worker.child.SystemApplicationClassLoaderWorker.call(SystemApplicationClassLoaderWorker.java:66)
at worker.org.gradle.process.internal.worker.GradleWorkerMain.run(GradleWorkerMain.java:69)
at worker.org.gradle.process.internal.worker.GradleWorkerMain.main(GradleWorkerMain.java:74)
Caused by: java.lang.OutOfMemoryError: GC overhead limit exceeded
at net.bytebuddy.description.type.TypeDescription$ForLoadedType.of(TypeDescription.java:8619)
at net.bytebuddy.description.method.MethodDescription$ForLoadedMethod.getDeclaringType(MethodDescription.java:1190)
at org.mockito.internal.creation.bytebuddy.MockMethodAdvice.isOverridden(MockMethodAdvice.java:199)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.isUnavailable(ConsumerNetworkClient.java:560)
at org.apache.kafka.clients.consumer.internals.Fetcher.isUnavailable(Fetcher.java:87)
at org.apache.kafka.clients.consumer.internals.AbstractFetch.prepareFetchRequests(AbstractFetch.java:427)
at org.apache.kafka.clients.consumer.internals.Fetcher.sendFetches(Fetcher.java:105)
at org.apache.kafka.clients.consumer.internals.FetcherTest.sendFetches(FetcherTest.java:246)
at org.apache.kafka.clients.consumer.internals.FetcherTest.testFetcherConcurrency(FetcherTest.java:2943)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:728)
```
the failed test is related to this PR. In the test case `testFetcherConcurrency`, it does not return correct `sessionTopicNames` so normally it should NOT see the correct response data. However, `FetchResponse#responseData` will return the cached data regardless of input, so it CAN get correct response data even though it pass empty `topicNames`. That is a good example of showing the potential bug :)
@m1a2st Could you copy the changes of this PR to another one, and please fix `testFetcherConcurrency` according to my comment. Also, please add new test for the change.
@johnnychhsu Sorry, I can't merge this PR as it causes the failed test. Please feel free to close this PR as @m1a2st will leverage this PR to complete it :)
--
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
Re: [PR] KAFKA-16684: Remove cache in responseData [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #15966:
URL: https://github.com/apache/kafka/pull/15966#issuecomment-2205221490
LGTM, I wll add test after this PR merge into trunk.
--
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
Re: [PR] KAFKA-16684: Remove cache in responseData [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #15966:
URL: https://github.com/apache/kafka/pull/15966#issuecomment-2153290523
@johnnychhsu This is a bug fix, so please add test for it
--
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
Re: [PR] KAFKA-16684: Remove cache in responseData [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #15966:
URL: https://github.com/apache/kafka/pull/15966#discussion_r1605998588
##########
clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java:
##########
@@ -99,28 +99,24 @@ public Errors error() {
}
public LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData(Map<Uuid, String> topicNames, short version) {
- if (responseData == null) {
- synchronized (this) {
- if (responseData == null) {
- // Assigning the lazy-initialized `responseData` in the last step
- // to avoid other threads accessing a half-initialized object.
- final LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseDataTmp =
- new LinkedHashMap<>();
- data.responses().forEach(topicResponse -> {
- String name;
- if (version < 13) {
- name = topicResponse.topic();
- } else {
- name = topicNames.get(topicResponse.topicId());
- }
- if (name != null) {
- topicResponse.partitions().forEach(partition ->
- responseDataTmp.put(new TopicPartition(name, partition.partitionIndex()), partition));
- }
- });
- responseData = responseDataTmp;
+ synchronized (this) {
+ // Assigning the lazy-initialized `responseData` in the last step
+ // to avoid other threads accessing a half-initialized object.
+ final LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseDataTmp =
+ new LinkedHashMap<>();
+ data.responses().forEach(topicResponse -> {
+ String name;
+ if (version < 13) {
+ name = topicResponse.topic();
+ } else {
+ name = topicNames.get(topicResponse.topicId());
}
- }
+ if (name != null) {
+ topicResponse.partitions().forEach(partition ->
+ responseDataTmp.put(new TopicPartition(name, partition.partitionIndex()), partition));
+ }
+ });
+ responseData = responseDataTmp;
Review Comment:
This is what I do concern! The `responseData` could be different if the input gets changed.
--
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
Re: [PR] KAFKA-16684: Remove cache in responseData [kafka]
Posted by "johnnychhsu (via GitHub)" <gi...@apache.org>.
johnnychhsu commented on code in PR #15966:
URL: https://github.com/apache/kafka/pull/15966#discussion_r1629471445
##########
clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java:
##########
@@ -99,28 +99,24 @@ public Errors error() {
}
public LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData(Map<Uuid, String> topicNames, short version) {
- if (responseData == null) {
- synchronized (this) {
- if (responseData == null) {
- // Assigning the lazy-initialized `responseData` in the last step
- // to avoid other threads accessing a half-initialized object.
- final LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseDataTmp =
- new LinkedHashMap<>();
- data.responses().forEach(topicResponse -> {
- String name;
- if (version < 13) {
- name = topicResponse.topic();
- } else {
- name = topicNames.get(topicResponse.topicId());
- }
- if (name != null) {
- topicResponse.partitions().forEach(partition ->
- responseDataTmp.put(new TopicPartition(name, partition.partitionIndex()), partition));
- }
- });
- responseData = responseDataTmp;
+ synchronized (this) {
+ // Assigning the lazy-initialized `responseData` in the last step
+ // to avoid other threads accessing a half-initialized object.
+ final LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseDataTmp =
+ new LinkedHashMap<>();
+ data.responses().forEach(topicResponse -> {
+ String name;
+ if (version < 13) {
+ name = topicResponse.topic();
+ } else {
+ name = topicNames.get(topicResponse.topicId());
}
- }
+ if (name != null) {
+ topicResponse.partitions().forEach(partition ->
+ responseDataTmp.put(new TopicPartition(name, partition.partitionIndex()), partition));
+ }
+ });
+ responseData = responseDataTmp;
Review Comment:
thanks for the review!
yes, the returned data should be calculated on the fly based on the input topic names
--
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
Re: [PR] KAFKA-16684: Remove cache in responseData [kafka]
Posted by "m1a2st (via GitHub)" <gi...@apache.org>.
m1a2st commented on PR #15966:
URL: https://github.com/apache/kafka/pull/15966#issuecomment-2207562526
@chia7712, Thanks for your comments, I will open new PR for this issue.
--
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
Re: [PR] KAFKA-16684: Remove cache in responseData [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #15966:
URL: https://github.com/apache/kafka/pull/15966#issuecomment-2208775816
close this PR @m1a2st will file another one
--
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
Re: [PR] KAFKA-16684: Remove cache in responseData [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 closed pull request #15966: KAFKA-16684: Remove cache in responseData
URL: https://github.com/apache/kafka/pull/15966
--
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