Skip to content

Commit 43e212a

Browse files
mjsaxmjsax
committed
rebased and fixed tests
1 parent 64c7f79 commit 43e212a

File tree

8 files changed

+104
-50
lines changed

8 files changed

+104
-50
lines changed

‎core/src/main/scala/kafka/server/KafkaApis.scala

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2636,13 +2636,13 @@ class KafkaApis(val requestChannel: RequestChannel,
26362636
}
26372637
}
26382638

2639-
private def streamsGroupVersion(): StreamsVersion = {
2639+
private def streamsVersion(): StreamsVersion = {
26402640
StreamsVersion.fromFeatureLevel(metadataCache.features.finalizedFeatures.getOrDefault(StreamsVersion.FEATURE_NAME, 0.toShort))
26412641
}
26422642

26432643
private def isStreamsGroupProtocolEnabled: Boolean = {
26442644
config.groupCoordinatorRebalanceProtocols.contains(Group.GroupType.STREAMS) &&
2645-
streamsGroupVersion().streamsGroupSupported
2645+
streamsVersion().streamsGroupSupported
26462646
}
26472647

26482648
def handleStreamsGroupHeartbeat(request: RequestChannel.Request): CompletableFuture[Unit] = {

‎core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

Lines changed: 89 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,7 @@ import org.apache.kafka.raft.QuorumConfig
8888
import org.apache.kafka.security.authorizer.AclEntry
8989
import org.apache.kafka.server.{ClientMetricsManager, SimpleApiVersionManager}
9090
import org.apache.kafka.server.authorizer.{Action, AuthorizationResult, Authorizer}
91-
import org.apache.kafka.server.common.{FeatureVersion, FinalizedFeatures, GroupVersion, KRaftVersion, MetadataVersion, RequestLocal, TransactionVersion}
91+
import org.apache.kafka.server.common.{FeatureVersion, FinalizedFeatures, GroupVersion, KRaftVersion, MetadataVersion, RequestLocal, StreamsVersion, TransactionVersion}
9292
import org.apache.kafka.server.config.{KRaftConfigs, ReplicationConfigs, ServerConfigs, ServerLogConfigs}
9393
import org.apache.kafka.server.metrics.ClientMetricsTestUtils
9494
import org.apache.kafka.server.share.{CachedSharePartition, ErroneousAndValidPartitionData, SharePartitionKey}
@@ -9902,7 +9902,11 @@ class KafkaApisTest extends Logging {
99029902

99039903
@Test
99049904
def testStreamsGroupHeartbeatRequest(): Unit = {
9905+
val features = mock(classOf[FinalizedFeatures])
9906+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
9907+
99059908
metadataCache = mock(classOf[KRaftMetadataCache])
9909+
when(metadataCache.features()).thenReturn(features)
99069910

99079911
val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group")
99089912

@@ -9913,9 +9917,7 @@ class KafkaApisTest extends Logging {
99139917
requestChannelRequest.context,
99149918
streamsGroupHeartbeatRequest
99159919
)).thenReturn(future)
9916-
kafkaApis = createKafkaApis(
9917-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
9918-
)
9920+
kafkaApis = createKafkaApis()
99199921
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
99209922

99219923
val streamsGroupHeartbeatResponse = new StreamsGroupHeartbeatResponseData()
@@ -9928,7 +9930,12 @@ class KafkaApisTest extends Logging {
99289930

99299931
@Test
99309932
def testStreamsGroupHeartbeatRequestWithAuthorizedTopology(): Unit = {
9933+
val features = mock(classOf[FinalizedFeatures])
9934+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
9935+
99319936
metadataCache = mock(classOf[KRaftMetadataCache])
9937+
when(metadataCache.features()).thenReturn(features)
9938+
99329939
val groupId = "group"
99339940
val fooTopicName = "foo"
99349941
val barTopicName = "bar"
@@ -9979,8 +9986,7 @@ class KafkaApisTest extends Logging {
99799986
streamsGroupHeartbeatRequest
99809987
)).thenReturn(future)
99819988
kafkaApis = createKafkaApis(
9982-
authorizer = Some(authorizer),
9983-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
9989+
authorizer = Some(authorizer)
99849990
)
99859991
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
99869992

@@ -9994,7 +10000,11 @@ class KafkaApisTest extends Logging {
999410000

999510001
@Test
999610002
def testStreamsGroupHeartbeatRequestFutureFailed(): Unit = {
10003+
val features = mock(classOf[FinalizedFeatures])
10004+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
10005+
999710006
metadataCache = mock(classOf[KRaftMetadataCache])
10007+
when(metadataCache.features()).thenReturn(features)
999810008

999910009
val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group")
1000010010

@@ -10005,9 +10015,7 @@ class KafkaApisTest extends Logging {
1000510015
requestChannelRequest.context,
1000610016
streamsGroupHeartbeatRequest
1000710017
)).thenReturn(future)
10008-
kafkaApis = createKafkaApis(
10009-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
10010-
)
10018+
kafkaApis = createKafkaApis()
1001110019
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
1001210020

1001310021
future.completeExceptionally(Errors.FENCED_MEMBER_EPOCH.exception)
@@ -10017,7 +10025,11 @@ class KafkaApisTest extends Logging {
1001710025

1001810026
@Test
1001910027
def testStreamsGroupHeartbeatRequestGroupAuthorizationFailed(): Unit = {
10028+
val features = mock(classOf[FinalizedFeatures])
10029+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
10030+
1002010031
metadataCache = mock(classOf[KRaftMetadataCache])
10032+
when(metadataCache.features()).thenReturn(features)
1002110033

1002210034
val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group")
1002310035

@@ -10027,8 +10039,7 @@ class KafkaApisTest extends Logging {
1002710039
when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
1002810040
.thenReturn(Seq(AuthorizationResult.DENIED).asJava)
1002910041
kafkaApis = createKafkaApis(
10030-
authorizer = Some(authorizer),
10031-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
10042+
authorizer = Some(authorizer)
1003210043
)
1003310044
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
1003410045

@@ -10038,7 +10049,12 @@ class KafkaApisTest extends Logging {
1003810049

1003910050
@Test
1004010051
def testStreamsGroupHeartbeatRequestTopicAuthorizationFailed(): Unit = {
10052+
val features = mock(classOf[FinalizedFeatures])
10053+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
10054+
1004110055
metadataCache = mock(classOf[KRaftMetadataCache])
10056+
when(metadataCache.features()).thenReturn(features)
10057+
1004210058
val groupId = "group"
1004310059
val fooTopicName = "foo"
1004410060
val barTopicName = "bar"
@@ -10079,8 +10095,7 @@ class KafkaApisTest extends Logging {
1007910095
}
1008010096

1008110097
kafkaApis = createKafkaApis(
10082-
authorizer = Some(authorizer),
10083-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
10098+
authorizer = Some(authorizer)
1008410099
)
1008510100
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
1008610101

@@ -10089,7 +10104,7 @@ class KafkaApisTest extends Logging {
1008910104
}
1009010105

1009110106
@Test
10092-
def testStreamsGroupHeartbeatRequestProtocolDisabled(): Unit = {
10107+
def testStreamsGroupHeartbeatRequestProtocolDisabledViaConfig(): Unit = {
1009310108
metadataCache = mock(classOf[KRaftMetadataCache])
1009410109

1009510110
val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group")
@@ -10105,9 +10120,32 @@ class KafkaApisTest extends Logging {
1010510120
assertEquals(Errors.UNSUPPORTED_VERSION.code, response.data.errorCode)
1010610121
}
1010710122

10123+
@Test
10124+
def testStreamsGroupHeartbeatRequestProtocolDisabledViaFeature(): Unit = {
10125+
val features = mock(classOf[FinalizedFeatures])
10126+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 0.toShort))
10127+
10128+
metadataCache = mock(classOf[KRaftMetadataCache])
10129+
when(metadataCache.features()).thenReturn(features)
10130+
10131+
val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group")
10132+
10133+
val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
10134+
10135+
kafkaApis = createKafkaApis()
10136+
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
10137+
10138+
val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
10139+
assertEquals(Errors.UNSUPPORTED_VERSION.code, response.data.errorCode)
10140+
}
10141+
1010810142
@Test
1010910143
def testStreamsGroupHeartbeatRequestInvalidTopicNames(): Unit = {
10144+
val features = mock(classOf[FinalizedFeatures])
10145+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
10146+
1011010147
metadataCache = mock(classOf[KRaftMetadataCache])
10148+
when(metadataCache.features()).thenReturn(features)
1011110149

1011210150
val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group").setTopology(
1011310151
new StreamsGroupHeartbeatRequestData.Topology()
@@ -10124,9 +10162,7 @@ class KafkaApisTest extends Logging {
1012410162

1012510163
val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
1012610164

10127-
kafkaApis = createKafkaApis(
10128-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
10129-
)
10165+
kafkaApis = createKafkaApis()
1013010166
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
1013110167

1013210168
val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
@@ -10136,7 +10172,11 @@ class KafkaApisTest extends Logging {
1013610172

1013710173
@Test
1013810174
def testStreamsGroupHeartbeatRequestInternalTopicNames(): Unit = {
10175+
val features = mock(classOf[FinalizedFeatures])
10176+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
10177+
1013910178
metadataCache = mock(classOf[KRaftMetadataCache])
10179+
when(metadataCache.features()).thenReturn(features)
1014010180

1014110181
val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group").setTopology(
1014210182
new StreamsGroupHeartbeatRequestData.Topology()
@@ -10152,9 +10192,7 @@ class KafkaApisTest extends Logging {
1015210192

1015310193
val requestChannelRequest = buildRequest(new StreamsGroupHeartbeatRequest.Builder(streamsGroupHeartbeatRequest, true).build())
1015410194

10155-
kafkaApis = createKafkaApis(
10156-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
10157-
)
10195+
kafkaApis = createKafkaApis()
1015810196
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
1015910197

1016010198
val response = verifyNoThrottling[StreamsGroupHeartbeatResponse](requestChannelRequest)
@@ -10164,7 +10202,11 @@ class KafkaApisTest extends Logging {
1016410202

1016510203
@Test
1016610204
def testStreamsGroupHeartbeatRequestWithInternalTopicsToCreate(): Unit = {
10205+
val features = mock(classOf[FinalizedFeatures])
10206+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
10207+
1016710208
metadataCache = mock(classOf[KRaftMetadataCache])
10209+
when(metadataCache.features()).thenReturn(features)
1016810210

1016910211
val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group");
1017010212

@@ -10176,9 +10218,7 @@ class KafkaApisTest extends Logging {
1017610218
streamsGroupHeartbeatRequest
1017710219
)).thenReturn(future)
1017810220

10179-
kafkaApis = createKafkaApis(
10180-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
10181-
)
10221+
kafkaApis = createKafkaApis()
1018210222
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
1018310223

1018410224
val missingTopics = Map("test" -> new CreatableTopic())
@@ -10193,7 +10233,11 @@ class KafkaApisTest extends Logging {
1019310233

1019410234
@Test
1019510235
def testStreamsGroupHeartbeatRequestWithInternalTopicsToCreateMissingCreateACL(): Unit = {
10236+
val features = mock(classOf[FinalizedFeatures])
10237+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
10238+
1019610239
metadataCache = mock(classOf[KRaftMetadataCache])
10240+
when(metadataCache.features()).thenReturn(features)
1019710241

1019810242
val streamsGroupHeartbeatRequest = new StreamsGroupHeartbeatRequestData().setGroupId("group");
1019910243

@@ -10219,8 +10263,7 @@ class KafkaApisTest extends Logging {
1021910263
}.asJava
1022010264
})
1022110265
kafkaApis = createKafkaApis(
10222-
authorizer = Some(authorizer),
10223-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
10266+
authorizer = Some(authorizer)
1022410267
)
1022510268
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
1022610269

@@ -10414,7 +10457,12 @@ class KafkaApisTest extends Logging {
1041410457
@ParameterizedTest
1041510458
@ValueSource(booleans = Array(true, false))
1041610459
def testStreamsGroupDescribe(includeAuthorizedOperations: Boolean): Unit = {
10460+
val features = mock(classOf[FinalizedFeatures])
10461+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
10462+
1041710463
metadataCache = mock(classOf[KRaftMetadataCache])
10464+
when(metadataCache.features()).thenReturn(features)
10465+
1041810466
val fooTopicName = "foo"
1041910467
val barTopicName = "bar"
1042010468

@@ -10429,9 +10477,7 @@ class KafkaApisTest extends Logging {
1042910477
any[RequestContext],
1043010478
any[util.List[String]]
1043110479
)).thenReturn(future)
10432-
kafkaApis = createKafkaApis(
10433-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
10434-
)
10480+
kafkaApis = createKafkaApis()
1043510481
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
1043610482

1043710483
val subtopology0 = new StreamsGroupDescribeResponseData.Subtopology()
@@ -10522,7 +10568,11 @@ class KafkaApisTest extends Logging {
1052210568

1052310569
@Test
1052410570
def testStreamsGroupDescribeAuthorizationFailed(): Unit = {
10571+
val features = mock(classOf[FinalizedFeatures])
10572+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
10573+
1052510574
metadataCache = mock(classOf[KRaftMetadataCache])
10575+
when(metadataCache.features()).thenReturn(features)
1052610576

1052710577
val streamsGroupDescribeRequestData = new StreamsGroupDescribeRequestData()
1052810578
streamsGroupDescribeRequestData.groupIds.add("group-id")
@@ -10539,8 +10589,7 @@ class KafkaApisTest extends Logging {
1053910589
)).thenReturn(future)
1054010590
future.complete(List().asJava)
1054110591
kafkaApis = createKafkaApis(
10542-
authorizer = Some(authorizer),
10543-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
10592+
authorizer = Some(authorizer)
1054410593
)
1054510594
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
1054610595

@@ -10550,7 +10599,11 @@ class KafkaApisTest extends Logging {
1055010599

1055110600
@Test
1055210601
def testStreamsGroupDescribeFutureFailed(): Unit = {
10602+
val features = mock(classOf[FinalizedFeatures])
10603+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
10604+
1055310605
metadataCache = mock(classOf[KRaftMetadataCache])
10606+
when(metadataCache.features()).thenReturn(features)
1055410607

1055510608
val streamsGroupDescribeRequestData = new StreamsGroupDescribeRequestData()
1055610609
streamsGroupDescribeRequestData.groupIds.add("group-id")
@@ -10561,9 +10614,7 @@ class KafkaApisTest extends Logging {
1056110614
any[RequestContext],
1056210615
any[util.List[String]]
1056310616
)).thenReturn(future)
10564-
kafkaApis = createKafkaApis(
10565-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
10566-
)
10617+
kafkaApis = createKafkaApis()
1056710618
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
1056810619

1056910620
future.completeExceptionally(Errors.FENCED_MEMBER_EPOCH.exception)
@@ -10578,7 +10629,11 @@ class KafkaApisTest extends Logging {
1057810629
val barTopicName = "bar"
1057910630
val errorMessage = "The described group uses topics that the client is not authorized to describe."
1058010631

10632+
val features = mock(classOf[FinalizedFeatures])
10633+
when(features.finalizedFeatures()).thenReturn(Collections.singletonMap(StreamsVersion.FEATURE_NAME, 1.toShort))
10634+
1058110635
metadataCache = mock(classOf[KRaftMetadataCache])
10636+
when(metadataCache.features()).thenReturn(features)
1058210637

1058310638
val groupIds = List("group-id-0", "group-id-1", "group-id-2").asJava
1058410639
val streamsGroupDescribeRequestData = new StreamsGroupDescribeRequestData()
@@ -10610,8 +10665,7 @@ class KafkaApisTest extends Logging {
1061010665
any[util.List[String]]
1061110666
)).thenReturn(future)
1061210667
kafkaApis = createKafkaApis(
10613-
authorizer = Some(authorizer),
10614-
overrideProperties = Map(GroupCoordinatorConfig.GROUP_COORDINATOR_REBALANCE_PROTOCOLS_CONFIG -> "classic,streams")
10668+
authorizer = Some(authorizer)
1061510669
)
1061610670
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
1061710671

‎core/src/test/scala/unit/kafka/tools/StorageToolTest.scala

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -325,7 +325,7 @@ Found problem:
325325
properties.putAll(defaultStaticQuorumProperties)
326326
properties.setProperty("log.dirs", availableDirs.mkString(","))
327327
assertEquals("Unsupported feature: non.existent.feature. Supported features are: " +
328-
"eligible.leader.replicas.version, group.version, kraft.version, share.version, streams.group.version, transaction.version",
328+
"eligible.leader.replicas.version, group.version, kraft.version, share.version, streams.version, transaction.version",
329329
assertThrows(classOf[FormatterException], () =>
330330
runFormatCommand(new ByteArrayOutputStream(), properties,
331331
Seq("--feature", "non.existent.feature=20"))).getMessage)

‎metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -396,7 +396,7 @@ public void testCannotDowngradeBeforeMinimumKraftVersion() {
396396
build();
397397
manager.replay(new FeatureLevelRecord().setName(MetadataVersion.FEATURE_NAME).setFeatureLevel(MetadataVersion.MINIMUM_VERSION.featureLevel()));
398398
assertEquals(ControllerResult.of(List.of(), new ApiError(Errors.INVALID_UPDATE_VERSION,
399-
"Invalid update version 6 for feature metadata.version. Local controller 0 only supports versions 7-29")),
399+
"Invalid update version 6 for feature metadata.version. Local controller 0 only supports versions 7-30")),
400400
manager.updateFeatures(
401401
Map.of(MetadataVersion.FEATURE_NAME, MetadataVersionTestUtils.IBP_3_3_IV2_FEATURE_LEVEL),
402402
Map.of(MetadataVersion.FEATURE_NAME, FeatureUpdate.UpgradeType.UNSAFE_DOWNGRADE),

‎metadata/src/test/java/org/apache/kafka/metadata/storage/FormatterTest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -368,7 +368,7 @@ public void testInvalidFeatureFlag() throws Exception {
368368
formatter1.formatter.setFeatureLevel("nonexistent.feature", (short) 1);
369369
assertEquals("Unsupported feature: nonexistent.feature. Supported features " +
370370
"are: eligible.leader.replicas.version, group.version, kraft.version, " +
371-
"share.version, streams.group.featrue, test.feature.version, transaction.version",
371+
"share.version, streams.version, test.feature.version, transaction.version",
372372
assertThrows(FormatterException.class,
373373
() -> formatter1.formatter.run()).
374374
getMessage());

‎server-common/src/main/java/org/apache/kafka/server/common/StreamsVersion.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ public static StreamsVersion fromFeatureLevel(short version) {
7777
return switch (version) {
7878
case 0 -> SV_0;
7979
case 1 -> SV_1;
80-
default -> throw new RuntimeException("Unknown streams group feature level: " + (int) version);
80+
default -> throw new RuntimeException("Unknown streams feature level: " + (int) version);
8181
};
8282
}
8383
}

‎test-common/test-common-internal-api/src/main/java/org/apache/kafka/common/test/api/ClusterTest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@
5252
String brokerListener() default DEFAULT_BROKER_LISTENER_NAME;
5353
SecurityProtocol controllerSecurityProtocol() default SecurityProtocol.PLAINTEXT;
5454
String controllerListener() default DEFAULT_CONTROLLER_LISTENER_NAME;
55-
MetadataVersion metadataVersion() default MetadataVersion.IBP_4_2_IV0;
55+
MetadataVersion metadataVersion() default MetadataVersion.IBP_4_2_IV1;
5656
ClusterConfigProperty[] serverProperties() default {};
5757
// users can add tags that they want to display in test
5858
String[] tags() default {};

0 commit comments

Comments
 (0)