KAFKA-19162: Topology metadata contains non-deterministically ordered topic configs (#19491)

Topology description sent to broker in KIP-1071 contains
non-deterministically ordered topic configs.  Since the topology is
compared to the groups topology upon joining we may run into
`INVALID_REQUEST: Topology updates are not supported yet` failures if
the topology sent by the  application does not match the group topology
due to different topic config order.

This PR ensures that topic configs are ordered, to avoid an
`INVALID_REQUEST` error.

Reviewers: Matthias J. Sax <[email protected]>

Author: lucasbru

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java

@@ -237,6 +237,7 @@ private static List<StreamsGroupHeartbeatRequestData.TopicInfo> getRepartitionTo
                     repartitionTopicInfo.topicConfigs().add(new StreamsGroupHeartbeatRequestData.KeyValue().setKey(k).setValue(v))
                 );
                 repartitionTopicsInfo.add(repartitionTopicInfo);
+                repartitionTopicInfo.topicConfigs().sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.KeyValue::key));
             }
             repartitionTopicsInfo.sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.TopicInfo::name));
             return repartitionTopicsInfo;
@@ -251,6 +252,7 @@ private static List<StreamsGroupHeartbeatRequestData.TopicInfo> getChangelogTopi
                 changelogTopic.getValue().topicConfigs().forEach((k, v) ->
                     changelogTopicInfo.topicConfigs().add(new StreamsGroupHeartbeatRequestData.KeyValue().setKey(k).setValue(v))
                 );
+                changelogTopicInfo.topicConfigs().sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.KeyValue::key));
                 changelogTopicsInfo.add(changelogTopicInfo);
             }
             changelogTopicsInfo.sort(Comparator.comparing(StreamsGroupHeartbeatRequestData.TopicInfo::name));

clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManagerTest.java

@@ -108,7 +108,7 @@ class StreamsGroupHeartbeatRequestManagerTest {
     private static final String REPARTITION_SOURCE_TOPIC_1 = "repartitionSourceTopic1";
     private static final String REPARTITION_SOURCE_TOPIC_2 = "repartitionSourceTopic2";
     private static final Map<String, StreamsRebalanceData.TopicInfo> REPARTITION_SOURCE_TOPICS = Map.of(
-        REPARTITION_SOURCE_TOPIC_1, new StreamsRebalanceData.TopicInfo(Optional.of(2), Optional.of((short) 1), Map.of("config1", "value1")),
+        REPARTITION_SOURCE_TOPIC_1, new StreamsRebalanceData.TopicInfo(Optional.of(2), Optional.of((short) 1), Map.of("config3", "value3", "config1", "value1")),
         REPARTITION_SOURCE_TOPIC_2, new StreamsRebalanceData.TopicInfo(Optional.of(3), Optional.of((short) 3), Collections.emptyMap())
     );
     private static final String CHANGELOG_TOPIC_1 = "changelogTopic1";
@@ -117,7 +117,7 @@ class StreamsGroupHeartbeatRequestManagerTest {
     private static final Map<String, StreamsRebalanceData.TopicInfo> CHANGELOG_TOPICS = Map.of(
         CHANGELOG_TOPIC_1, new StreamsRebalanceData.TopicInfo(Optional.empty(), Optional.of((short) 1), Map.of()),
         CHANGELOG_TOPIC_2, new StreamsRebalanceData.TopicInfo(Optional.empty(), Optional.of((short) 2), Map.of()),
-        CHANGELOG_TOPIC_3, new StreamsRebalanceData.TopicInfo(Optional.empty(), Optional.of((short) 3), Map.of("config2", "value2"))
+        CHANGELOG_TOPIC_3, new StreamsRebalanceData.TopicInfo(Optional.empty(), Optional.of((short) 3), Map.of("config4", "value4", "config2", "value2"))
     );
     private static final Collection<Set<String>> COPARTITION_GROUP = Set.of(
         Set.of(SOURCE_TOPIC_1, REPARTITION_SOURCE_TOPIC_2),
@@ -664,6 +664,7 @@ public void testBuildingHeartbeatRequestTopologySentWhenJoining(final MemberStat
             assertEquals(repartitionTopic.numPartitions().get(), topicInfo.partitions());
             assertEquals(repartitionTopic.replicationFactor().get(), topicInfo.replicationFactor());
             assertEquals(repartitionTopic.topicConfigs().size(), topicInfo.topicConfigs().size());
+            assertTrue(isSorted(topicInfo.topicConfigs(), Comparator.comparing(StreamsGroupHeartbeatRequestData.KeyValue::key)));
         });
         assertEquals(CHANGELOG_TOPICS.size(), subtopology1.stateChangelogTopics().size());
         subtopology1.stateChangelogTopics().forEach(topicInfo -> {
@@ -672,6 +673,7 @@ public void testBuildingHeartbeatRequestTopologySentWhenJoining(final MemberStat
             final StreamsRebalanceData.TopicInfo changelogTopic = CHANGELOG_TOPICS.get(topicInfo.name());
             assertEquals(changelogTopic.replicationFactor().get(), topicInfo.replicationFactor());
             assertEquals(changelogTopic.topicConfigs().size(), topicInfo.topicConfigs().size());
+            assertTrue(isSorted(topicInfo.topicConfigs(), Comparator.comparing(StreamsGroupHeartbeatRequestData.KeyValue::key)));
         });
         assertEquals(2, subtopology1.copartitionGroups().size());
         final StreamsGroupHeartbeatRequestData.CopartitionGroup expectedCopartitionGroupData1 =
@@ -701,6 +703,15 @@ public void testBuildingHeartbeatRequestTopologySentWhenJoining(final MemberStat
         assertNull(nonJoiningRequestData.topology());
     }
 
+    private <V> boolean isSorted(List<V> collection, Comparator<V> comparator) {
+        for (int i = 1; i < collection.size(); i++) {
+            if (comparator.compare(collection.get(i - 1), collection.get(i)) > 0) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     @ParameterizedTest
     @MethodSource("provideNonJoiningStates")
     public void testBuildingHeartbeatRequestRebalanceTimeoutSentWhenJoining(final MemberState memberState) {