KAFKA-15771: fix ProduceRequest#partitionSizes() to make it an atomic operation

[fxbing]

Encountered a concurrency issue in method ProduceRequest#partitionSizes() while developing with Kafka. When both Thread 1 and Thread 2 concurrently call method ProduceRequest#partitionSizes(), Thread 2 may receive an incomplete or empty result if Thread 1 is still in the process of initializing partitionSizes. This is an incorrect state. the code to ensure that Thread 2 obtains the final state rather than an intermediate one.

[vamossagar12]

Thanks for the PR @fxbing . I am not totally clear about what we are trying to solve using this though. Could you please elaborate in the PR description section? Is there a race condition that you have identified?

[fxbing]

Thanks for the PR @fxbing . I am not totally clear about what we are trying to solve using this though. Could you please elaborate in the PR description section? Is there a race condition that you have identified?

@vamossagar12 I added some instructions, please see if there are other questions

[hudeqi]

Thanks for this PR! @fxbing. I think this deserves a jira to track. And can you add a unit test?

[fxbing]

Thanks for this PR! @fxbing. I think this deserves a jira to track. And can you add a unit test?

@hudeqi Thanks for the comments, I've added the jira issue.
But regarding unit testing, it is not easy to conduct non-intrusive unit testing accurately. What I can think of is to ensure consistent results through concurrent calls by many threads, but even in the problematic version, there is a certain probability of successful execution.
Do you want to add such unit tests? Any other better suggestions?

[vamossagar12]

Thanks for the explanation @fxbing and for creating the JIRA ticket. It makes sense now. LGTM. Regarding tests, I agree it might be difficult to write a test which will simulate this race condition always. I am ok to not have such a test.

[vamossagar12]

@chia7712 , do you ming taking a look at this one plz?

[fxbing]

@vamossagar12 @hudeqi Can this PR be merged?

[vamossagar12]

@fxbing , Both me and @hudeqi don't have access to merge PRs. We will have to wait for somebody who has access. I had pinged @chia7712 who has access and expertise on this part of code.

[divijvaidya]

Thank you for the change. Looks good! I will backport it to 3.5 and 3.6 as well.

[divijvaidya]

Test failures and build failure is unrelated:

[Build / JDK 21 and Scala 2.13 / org.apache.kafka.connect.integration.OffsetsApiIntegrationTest.testAlterSinkConnectorOffsetsZombieSinkTasks](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14674/3/testReport/junit/org.apache.kafka.connect.integration/OffsetsApiIntegrationTest/Build___JDK_21_and_Scala_2_13___testAlterSinkConnectorOffsetsZombieSinkTasks/)
[Build / JDK 21 and Scala 2.13 / kafka.coordinator.transaction.ProducerIdManagerTest.[1] error=UNKNOWN_SERVER_ERROR](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14674/3/testReport/junit/kafka.coordinator.transaction/ProducerIdManagerTest/Build___JDK_21_and_Scala_2_13____1__error_UNKNOWN_SERVER_ERROR/)
[Build / JDK 21 and Scala 2.13 / kafka.log.remote.RemoteIndexCacheTest.testClose()](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14674/3/testReport/junit/kafka.log.remote/RemoteIndexCacheTest/Build___JDK_21_and_Scala_2_13___testClose__/)
[Build / JDK 8 and Scala 2.12 / kafka.integration.MetricsDuringTopicCreationDeletionTest.testMetricsDuringTopicCreateDelete(String).quorum=zk](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14674/3/testReport/junit/kafka.integration/MetricsDuringTopicCreationDeletionTest/Build___JDK_8_and_Scala_2_12___testMetricsDuringTopicCreateDelete_String__quorum_zk/)
[Build / JDK 8 and Scala 2.12 / kafka.server.MetadataRequestBetweenDifferentIbpTest.testUnknownTopicId()](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14674/3/testReport/junit/kafka.server/MetadataRequestBetweenDifferentIbpTest/Build___JDK_8_and_Scala_2_12___testUnknownTopicId__/)
[Build / JDK 8 and Scala 2.12 / org.apache.kafka.streams.integration.AdjustStreamThreadCountTest.shouldAddStreamThread()](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14674/3/testReport/junit/org.apache.kafka.streams.integration/AdjustStreamThreadCountTest/Build___JDK_8_and_Scala_2_12___shouldAddStreamThread__/)
[Build / JDK 17 and Scala 2.13 / org.apache.kafka.connect.integration.OffsetsApiIntegrationTest.testGetSinkConnectorOffsets](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14674/3/testReport/junit/org.apache.kafka.connect.integration/OffsetsApiIntegrationTest/Build___JDK_17_and_Scala_2_13___testGetSinkConnectorOffsets/)
[Build / JDK 17 and Scala 2.13 / integration.kafka.server.FetchFromFollowerIntegrationTest.testRackAwareRangeAssignor(String).quorum=zk](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14674/3/testReport/junit/integration.kafka.server/FetchFromFollowerIntegrationTest/Build___JDK_17_and_Scala_2_13___testRackAwareRangeAssignor_String__quorum_zk/)
[Build / JDK 17 and Scala 2.13 / kafka.api.ConsumerBounceTest.testClose()](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14674/3/testReport/junit/kafka.api/ConsumerBounceTest/Build___JDK_17_and_Scala_2_13___testClose__/)
[Build / JDK 17 and Scala 2.13 / kafka.api.DelegationTokenEndToEndAuthorizationWithOwnerTest.testProduceConsumeViaSubscribe(String).quorum=kraft](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14674/3/testReport/junit/kafka.api/DelegationTokenEndToEndAuthorizationWithOwnerTest/Build___JDK_17_and_Scala_2_13___testProduceConsumeViaSubscribe_String__quorum_kraft/)