MINOR: Avoid a lambda allocation per lookup in CachedTopicResolver (#23710)

`CachedTopicResolver.id` and `CachedTopicResolver.name` pass a lambda
capturing the key to `HashMap.computeIfAbsent`, so every lookup
allocates one, cache hits included. The JIT does not eliminate it:
`HashMap.computeIfAbsent` is above C2's inlining limit, so the lambda
escapes. `TopicIds` calls `id` for every topic it iterates and `name` in
`contains`, so an assignor walking the subscriptions of a heterogeneous
group allocates one lambda per member and subscribed topic.

This patch reads the cache with `get` and only looks up the image and
calls `put` on a miss. As before, an unknown topic resolves to `null`
and is not cached. It also adds `topicId` and `topicName` to
`CoordinatorMetadataImage`, which look up a topic without creating an
`Optional` and a `TopicMetadata`, and uses them in both resolvers.

Bytes allocated per assignment (`gc.alloc.rate.norm`) with heterogeneous
subscriptions, before and after.

`ConsumerAssignorBenchmark`, 5000 members, 20000 partitions, `FULL`:

| Assignor | Subscription           | Topics | Before     | After    |
|----------|------------------------|--------|------------|----------| |
UNIFORM  | HETEROGENEOUS_DISJOINT | 500    | 47.3 MB    | 35.2 MB  | |
UNIFORM  | HETEROGENEOUS_NESTED   | 500    | 126.2 MB   | 90.1 MB  | |
UNIFORM  | HETEROGENEOUS_DISJOINT | 5000   | 422.7 MB   | 302.8 MB | |
UNIFORM  | HETEROGENEOUS_NESTED   | 5000   | 1,211.9 MB | 852.2 MB | |
RANGE    | HETEROGENEOUS_DISJOINT | 500    | 38.0 MB    | 14.4 MB  | |
RANGE    | HETEROGENEOUS_NESTED   | 500    | 110.1 MB   | 38.4 MB  | |
RANGE    | HETEROGENEOUS_DISJOINT | 5000   | 362.6 MB   | 123.2 MB | |
RANGE    | HETEROGENEOUS_NESTED   | 5000   | 1,083.0 MB | 363.3 MB |

`TargetAssignmentBuilderBenchmark`, 10 partitions per member,
`HETEROGENEOUS`:

| Members | Topics | Before   | After    |
|---------|--------|----------|----------| | 1000    | 100    | 2.4 MB |
1.7 MB   | | 1000    | 1000   | 17.3 MB  | 12.2 MB  | | 10000   | 100 |
22.1 MB  | 14.9 MB  | | 10000   | 1000   | 162.5 MB | 112.0 MB |

The `STABLE` event gives similar reductions. Homogeneous subscriptions
resolve the topics of a single member, so the saving there is at most
one lambda per topic.

Reviewers: Sean Quah <squah@confluent.io>, Chia-Ping Tsai
 <chia7712@gmail.com>

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorMetadataImage.java

@@ -37,6 +37,16 @@ public interface CoordinatorMetadataImage {
 
     Optional<TopicMetadata> topicMetadata(Uuid topicId);
 
+    /**
+     * @return The id of the topic, or null if the topic does not exist.
+     */
+    Uuid topicId(String topicName);
+
+    /**
+     * @return The name of the topic, or null if the topic does not exist.
+     */
+    String topicName(Uuid topicId);
+
     CoordinatorMetadataDelta emptyDelta();
 
     long version();
@@ -79,6 +89,16 @@ public interface CoordinatorMetadataImage {
                 return Optional.empty();
             }
 
+            @Override
+            public Uuid topicId(String topicName) {
+                return null;
+            }
+
+            @Override
+            public String topicName(Uuid topicId) {
+                return null;
+            }
+
             @Override
             public CoordinatorMetadataDelta emptyDelta() {
                 return CoordinatorMetadataDelta.EMPTY;

coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/KRaftCoordinatorMetadataImage.java

@@ -74,6 +74,18 @@ public class KRaftCoordinatorMetadataImage implements CoordinatorMetadataImage {
         return Optional.of(new KraftTopicMetadata(topicImage, clusterImage));
     }
 
+    @Override
+    public Uuid topicId(String topicName) {
+        TopicImage topicImage = metadataImage.topics().getTopic(topicName);
+        return topicImage == null ? null : topicImage.id();
+    }
+
+    @Override
+    public String topicName(Uuid topicId) {
+        TopicImage topicImage = metadataImage.topics().getTopic(topicId);
+        return topicImage == null ? null : topicImage.name();
+    }
+
     @Override
     public CoordinatorMetadataDelta emptyDelta() {
         // Note: supportedConfigChecker is not set because CoordinatorMetadataDelta only exposes topic-related methods.

coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/KRaftCoordinatorMetadataImageTest.java

@@ -28,6 +28,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.fail;
 
 class KRaftCoordinatorMetadataImageTest {
@@ -140,4 +141,31 @@ class KRaftCoordinatorMetadataImageTest {
         assertEquals(coordinatorMetadataImage.hashCode(), coordinatorMetadataImageCopy.hashCode());
         assertNotEquals(coordinatorMetadataImage.hashCode(), coordinatorMetadataImage2.hashCode());
     }
+
+    @Test
+    public void testTopicIdAndTopicName() {
+        Uuid topicId = Uuid.randomUuid();
+        String topicName = "test-topic";
+        Uuid noPartitionTopicId = Uuid.randomUuid();
+        String noPartitionTopic = "no-partition-topic";
+
+        KRaftCoordinatorMetadataImage image = new KRaftCoordinatorMetadataImage(new MetadataImageBuilder()
+            .addTopic(topicId, topicName, 2)
+            .addTopic(noPartitionTopicId, noPartitionTopic, 0)
+            .build());
+
+        assertEquals(topicId, image.topicId(topicName));
+        assertEquals(topicName, image.topicName(topicId));
+        assertEquals(noPartitionTopicId, image.topicId(noPartitionTopic));
+        assertEquals(noPartitionTopic, image.topicName(noPartitionTopicId));
+
+        assertNull(image.topicId("unknown-topic"));
+        assertNull(image.topicName(Uuid.randomUuid()));
+    }
+
+    @Test
+    public void testTopicIdAndTopicNameOfEmptyImage() {
+        assertNull(CoordinatorMetadataImage.EMPTY.topicId("test-topic"));
+        assertNull(CoordinatorMetadataImage.EMPTY.topicName(Uuid.randomUuid()));
+    }
 }

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/TopicIds.java

@@ -79,12 +79,12 @@ public class TopicIds implements Set<Uuid> {
 
         @Override
         public String name(Uuid id) {
-            return image.topicMetadata(id).map(CoordinatorMetadataImage.TopicMetadata::name).orElse(null);
+            return image.topicName(id);
         }
 
         @Override
         public Uuid id(String name) {
-            return image.topicMetadata(name).map(CoordinatorMetadataImage.TopicMetadata::id).orElse(null);
+            return image.topicId(name);
         }
 
         @Override
@@ -122,12 +122,30 @@ public class TopicIds implements Set<Uuid> {
 
         @Override
         public String name(Uuid id) {
-            return topicNames.computeIfAbsent(id, __ -> image.topicMetadata(id).map(CoordinatorMetadataImage.TopicMetadata::name).orElse(null));
+            // Uses get and put rather than computeIfAbsent, whose lambda would capture the id and
+            // be allocated on every call, cache hits included.
+            String name = topicNames.get(id);
+            if (name == null) {
+                name = image.topicName(id);
+                if (name != null) {
+                    topicNames.put(id, name);
+                }
+            }
+            return name;
         }
 
         @Override
         public Uuid id(String name) {
-            return topicIds.computeIfAbsent(name, __ -> image.topicMetadata(name).map(CoordinatorMetadataImage.TopicMetadata::id).orElse(null));
+            // Uses get and put rather than computeIfAbsent, whose lambda would capture the name
+            // and be allocated on every call, cache hits included.
+            Uuid id = topicIds.get(name);
+            if (id == null) {
+                id = image.topicId(name);
+                if (id != null) {
+                    topicIds.put(name, id);
+                }
+            }
+            return id;
         }
 
         @Override

group-coordinator/src/test/java/org/apache/kafka/coordinator/group/modern/TopicIdsTest.java

@@ -32,6 +32,9 @@ import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 public class TopicIdsTest {
 
@@ -231,4 +234,57 @@ public class TopicIdsTest {
 
         assertEquals(topicIds1, topicIds2);
     }
+
+    @Test
+    public void testCachedTopicResolverId() {
+        var fooUuid = Uuid.randomUuid();
+        var metadataImage = spy(new KRaftCoordinatorMetadataImage(
+            new MetadataImageBuilder()
+                .addTopic(fooUuid, "foo", 3)
+                .build()
+        ));
+        var resolver = new TopicIds.CachedTopicResolver(metadataImage);
+
+        // A known topic is looked up in the image once, then served from the cache.
+        assertEquals(fooUuid, resolver.id("foo"));
+        assertEquals(fooUuid, resolver.id("foo"));
+        verify(metadataImage, times(1)).topicId("foo");
+
+        // An unknown topic is not cached, so it is looked up in the image every time.
+        assertNull(resolver.id("bar"));
+        assertNull(resolver.id("bar"));
+        verify(metadataImage, times(2)).topicId("bar");
+
+        // A cleared cache looks up the known topic in the image again.
+        resolver.clear();
+        assertEquals(fooUuid, resolver.id("foo"));
+        verify(metadataImage, times(2)).topicId("foo");
+    }
+
+    @Test
+    public void testCachedTopicResolverName() {
+        var fooUuid = Uuid.randomUuid();
+        var barUuid = Uuid.randomUuid();
+        var metadataImage = spy(new KRaftCoordinatorMetadataImage(
+            new MetadataImageBuilder()
+                .addTopic(fooUuid, "foo", 3)
+                .build()
+        ));
+        var resolver = new TopicIds.CachedTopicResolver(metadataImage);
+
+        // A known topic is looked up in the image once, then served from the cache.
+        assertEquals("foo", resolver.name(fooUuid));
+        assertEquals("foo", resolver.name(fooUuid));
+        verify(metadataImage, times(1)).topicName(fooUuid);
+
+        // An unknown topic is not cached, so it is looked up in the image every time.
+        assertNull(resolver.name(barUuid));
+        assertNull(resolver.name(barUuid));
+        verify(metadataImage, times(2)).topicName(barUuid);
+
+        // A cleared cache looks up the known topic in the image again.
+        resolver.clear();
+        assertEquals("foo", resolver.name(fooUuid));
+        verify(metadataImage, times(2)).topicName(fooUuid);
+    }
 }