KAFKA-17336 Add IT to make sure the production MV does not use unstable version of LIST_OFFSET #16893
KAFKA-17336 Add IT to make sure the production MV does not use unstable version of LIST_OFFSET #16893chia7712 merged 6 commits intoapache:trunkfrom
Conversation
|
If ListOffsetRequest latestVersionUnstable is true, the
|
| @ParameterizedTest | ||
| @ValueSource(strings = Array("zk", "kraft")) | ||
| def testResponseIncludesLeaderEpoch(quorum: String): Unit = { | ||
| def testResponseIncludesLeaderEpochWithUnstableAPIs(quorum: String): Unit = { |
There was a problem hiding this comment.
Could you please add comments for this test? this comment is a good reference (#16841 (comment))
|
|
||
| @BeforeEach | ||
| override def setUp(testInfo: TestInfo): Unit = { | ||
| enableUnstableVersions = !testInfo.getDisplayName.contains("testResponseIncludesLeaderEpochWithUnstableAPIs") |
There was a problem hiding this comment.
If the test name contains "UnstableAPI", we set enableUnstableVersions to false. Setting it to true will be more intuitive.
There was a problem hiding this comment.
Yes, It make sence when test istestResponseIncludesLeaderEpochWithStableAPIs set enableUnstableVersions to false is more better.
| testResponseIncludesLeaderEpoch() | ||
| } | ||
|
|
||
| def testResponseIncludesLeaderEpoch(): Unit = { |
There was a problem hiding this comment.
I am not sure if this truly captures the problem. The test seems to verify that the broker can accept the latest version of ListOffset when enableUnstableVersions is set to false. This means that we could never introduce an unstable version of ListOffset in the future. When introducing a new version of ListOffset, it's ok to make it unstable, as long as the stable MV doesn't use it for inter broker communication.
There was a problem hiding this comment.
Maybe we can have a IT used to make sure MV does not return "unstable" version for inter broker communication. for example:
@ClusterTests({
@ClusterTest(serverProperties = {@ClusterConfigProperty(key = ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, value = "false")}),
@ClusterTest(serverProperties = {@ClusterConfigProperty(key = ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, value = "true")})
})
public void test(ClusterInstance clusterInstance) {
clusterInstance.brokers().values().forEach(b -> {
MetadataVersion mv = b.metadataCache().metadataVersion();
Assertions.assertTrue(mv.listOffsetRequestVersion() <= ApiKeys.LIST_OFFSETS.latestVersion(false));
Assertions.assertTrue(mv.fetchRequestVersion() <= ApiKeys.FETCH.latestVersion(false));
Assertions.assertTrue(mv.offsetForLeaderEpochRequestVersion() <= ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion(false));
});
}There was a problem hiding this comment.
@chia7712 : Good idea. We probably don't need an IT. We could just use the latest production MV like the following.
public void test() {
MetadataVersion mv = MetadataVersion.latestProduction()
Assertions.assertTrue(mv.listOffsetRequestVersion() <= ApiKeys.LIST_OFFSETS.latestVersion(false));
Assertions.assertTrue(mv.fetchRequestVersion() <= ApiKeys.FETCH.latestVersion(false));
Assertions.assertTrue(mv.offsetForLeaderEpochRequestVersion() <= ApiKeys.OFFSET_FOR_LEADER_EPOCH.latestVersion(false));
});
}
There was a problem hiding this comment.
TransactionMarkerChannelManager sets the version for WriteTxnMarkersRequest based on MV in the following code in TransactionMarkerChannelManager. It would be useful to move that logic to MetadataVersion and add a similar test on unstable version. This could be done in a separate PR if it requires more work.
private val writeTxnMarkersRequestVersion: Short =
if (config.interBrokerProtocolVersion.isAtLeast(IBP_2_8_IV0)) 1
else 0
There was a problem hiding this comment.
This could be done in a separate PR if it requires more work.
I feel following small changes can address it, so this PR should be able to include it. @m1a2st WDYT?
public short writeTxnMarkersRequestVersion() {
if (isAtLeast(IBP_2_8_IV0)) {
return 1;
} else {
return 0;
}
}There was a problem hiding this comment.
Yes, I will address these comments in this PR.
| val requestCompletionHandler = new TransactionMarkerRequestCompletionHandler(node.id, txnStateManager, this, entries) | ||
| val request = new WriteTxnMarkersRequest.Builder(writeTxnMarkersRequestVersion, markersToSend) | ||
| val request = new WriteTxnMarkersRequest.Builder( | ||
| config.interBrokerProtocolVersion.writeTxnMarkersRequestVersion(), markersToSend |
There was a problem hiding this comment.
It seems that we should get the MV from metadataCache.metadataVersion() instead of config.interBrokerProtocolVersion since MV can be set dynamically. @jolshan : What do you think?
There was a problem hiding this comment.
Yes. Config.interbrokerProtocolVersion is not the reliable way to get MV.
There was a problem hiding this comment.
config.interBrokerProtocolVersion is used in code base. in zk mode, metadataCache.metadataVersion is static config. By contrast, kraft has dynamic MV. Hence, we should apply that change (config.interBrokerProtocolVersion -> metadataCache.metadataVersion) to code base. BTW, that will be a large PR I guess @junrao @jolshan WDYT?
There was a problem hiding this comment.
Taking a quick look, it doesn't look like it is used in too many places, but can definitely be done as a followup.
There was a problem hiding this comment.
but can definitely be done as a followup.
@m1a2st WDYT? Sorry that I'm in traveling so can't check the code usages
There was a problem hiding this comment.
Yes, If we want to change config.interBrokerProtocolVersion -> metadataCache.metadataVersion, there are my finding, but exclude test:
- https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/cluster/Partition.scala#L143
- https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala#L1799
- https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaApis.scala#L458
- https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaApis.scala#L1802
- https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaApis.scala#L1832
- https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaApis.scala#L1901
- https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaApis.scala#L2532
- https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaApis.scala#L2533
- https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/ConfigHandler.scala#L174
- https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/ConfigHandler.scala#L169
There are some use in KafkaServer.scala, but I think it we should not change
@chia7712, PTAL, Thanks you
There was a problem hiding this comment.
@m1a2st thanks for sharing. It seems we can address it in this PR as it is not used in too many places.
| } | ||
|
|
||
| @Test | ||
| public void testLastMetadataProductionDontReturnUnstableVersion() { |
There was a problem hiding this comment.
Perhaps add a comment like the following.
The broker picks the version for a few inter broker RPCs based on the metadata version, instead of the supported version from ApiResponse. We need to make sure that the latest production MV doesn't accidentally depend on an unstable request version.
testLastMetadataProductionDontReturnUnstableVersion => testProductionMetadataDontUseUnstableApiVersion ?
|
Hello @chia7712, If modify KafkaApis, some tests in KafkaApisTest will be fail.
|
| def ensureInterBrokerVersion(version: MetadataVersion): Unit = { | ||
| if (config.interBrokerProtocolVersion.isLessThan(version)) | ||
| throw new UnsupportedVersionException(s"inter.broker.protocol.version: ${config.interBrokerProtocolVersion} is less than the required version: ${version}") | ||
| if (replicaManager.metadataCache.metadataVersion().isLessThan(version)) |
There was a problem hiding this comment.
why taking metadataCache from replicaManager? KafkaApis has metadataCache already
| } | ||
| } | ||
|
|
||
| when(replicaManager.metadataCache) |
There was a problem hiding this comment.
All we need to do is to create MetadataCache according to MetadataVersion used by testing. for example:
@Test
def shouldThrowUnsupportedVersionExceptionOnHandleAddOffsetToTxnRequestWhenInterBrokerProtocolNotSupported(): Unit = {
metadataCache = MetadataCache.zkMetadataCache(brokerId, IBP_0_10_2_IV0)
brokerEpochManager = new ZkBrokerEpochManager(metadataCache, controller, None)
kafkaApis = createKafkaApis(IBP_0_10_2_IV0)
assertThrows(classOf[UnsupportedVersionException],
() => kafkaApis.handleAddOffsetsToTxnRequest(null, RequestLocal.withThreadConfinedCaching))
}
m1a2st
force-pushed
the
KAFKA-17336
branch
3 times, most recently
from
0af86ec to
9abcf35
Compare
| if (config.interBrokerProtocolVersion.isLessThan(version)) | ||
| throw new UnsupportedVersionException(s"inter.broker.protocol.version: ${config.interBrokerProtocolVersion} is less than the required version: ${version}") | ||
| if (metadataCache.metadataVersion().isLessThan(version)) | ||
| throw new UnsupportedVersionException(s"inter.broker.protocol.version: ${metadataCache.metadataVersion()} is less than the required version: ${version}") |
There was a problem hiding this comment.
Should we change inter.broker.protocol.version accordingly?
There was a problem hiding this comment.
Thanks for your comment, I will modify this. Should I use the metadata.version is clear enough?
tests with JDK 11 are not completed. @m1a2st Could you please rebase code to run CI again? the other failed tests pass on my local. |
oh, I run the tests with incorrect PR.
|
|
These tests passed on my local exclude |
I will take a look later. |
|
|
|
The other failed tests pass on my local. https://issues.apache.org/jira/browse/KAFKA-16636 |
…le version of LIST_OFFSET (#16893) - due to the server config UNSTABLE_API_VERSIONS_ENABLE_CONFIG is true, so we can't test the scenario of ListOffsetsRequest is unstable version. We want to test this case in this PR - get the MV from metadataCache.metadataVersion() instead of config.interBrokerProtocolVersion since MV can be set dynamically. Reviewers: Jun Rao <junrao@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
…e unstable version of LIST_OFFSET (apache#16893)" This reverts commit a3aa637.
…le version of LIST_OFFSET (apache#16893) - due to the server config UNSTABLE_API_VERSIONS_ENABLE_CONFIG is true, so we can't test the scenario of ListOffsetsRequest is unstable version. We want to test this case in this PR - get the MV from metadataCache.metadataVersion() instead of config.interBrokerProtocolVersion since MV can be set dynamically. Reviewers: Jun Rao <junrao@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
…le version of LIST_OFFSET (apache#16893) - due to the server config UNSTABLE_API_VERSIONS_ENABLE_CONFIG is true, so we can't test the scenario of ListOffsetsRequest is unstable version. We want to test this case in this PR - get the MV from metadataCache.metadataVersion() instead of config.interBrokerProtocolVersion since MV can be set dynamically. Reviewers: Jun Rao <junrao@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>
4 participants
Jira: https://issues.apache.org/jira/browse/KAFKA-17336
Due to the server config
UNSTABLE_API_VERSIONS_ENABLE_CONFIGis true, so we can't test the scenario of ListOffsetsRequest is unstable version. We want to test this case in this PRCommitter Checklist (excluded from commit message)