diff --git a/clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java b/clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java
index 027648f3edfa8..09d861e69efbe 100644
--- a/clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java
+++ b/clients/src/main/java/org/apache/kafka/common/TopicIdPartition.java
@@ -27,16 +27,27 @@ public class TopicIdPartition {
     private final Uuid topicId;
     private final TopicPartition topicPartition;
 
+    /**
+     * Create an instance with the provided parameters.
+     *
+     * @param topicId the topic id
+     * @param topicPartition the topic partition
+     */
     public TopicIdPartition(Uuid topicId, TopicPartition topicPartition) {
         this.topicId = Objects.requireNonNull(topicId, "topicId can not be null");
         this.topicPartition = Objects.requireNonNull(topicPartition, "topicPartition can not be null");
     }
 
-    public TopicIdPartition(String topic, Uuid topicId, int partition) {
+    /**
+     * Create an instance with the provided parameters.
+     *
+     * @param topicId the topic id
+     * @param partition the partition id
+     * @param topic the topic name or null
+     */
+    public TopicIdPartition(Uuid topicId, int partition, String topic) {
         this.topicId = Objects.requireNonNull(topicId, "topicId can not be null");
-        this.topicPartition = new TopicPartition(
-            Objects.requireNonNull(topic, "topic can not be null"),
-            partition);
+        this.topicPartition = new TopicPartition(topic, partition);
     }
 
     /**
@@ -47,7 +58,7 @@ public Uuid topicId() {
     }
 
     /**
-     * @return the topic name.
+     * @return the topic name or null if it is unknown.
      */
     public String topic() {
         return topicPartition.topic();
diff --git a/clients/src/test/java/org/apache/kafka/common/TopicIdPartitionTest.java b/clients/src/test/java/org/apache/kafka/common/TopicIdPartitionTest.java
index 08de64f8cb127..d39481c5b37d1 100644
--- a/clients/src/test/java/org/apache/kafka/common/TopicIdPartitionTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/TopicIdPartitionTest.java
@@ -30,21 +30,31 @@ class TopicIdPartitionTest {
     private final int partition1 = 1;
     private final TopicPartition topicPartition0 = new TopicPartition(topicName0, partition1);
     private final TopicIdPartition topicIdPartition0 = new TopicIdPartition(topicId0, topicPartition0);
-    private final TopicIdPartition topicIdPartition1 = new TopicIdPartition(topicName0, topicId0,
-        partition1);
+    private final TopicIdPartition topicIdPartition1 = new TopicIdPartition(topicId0,
+        partition1, topicName0);
+
+    private final TopicIdPartition topicIdPartitionWithNullTopic0 = new TopicIdPartition(topicId0,
+        partition1, null);
+    private final TopicIdPartition topicIdPartitionWithNullTopic1 = new TopicIdPartition(topicId0,
+        new TopicPartition(null, partition1));
 
     private final Uuid topicId1 = new Uuid(7759286116672424028L, -5081215629859775948L);
     private final String topicName1 = "another_topic_name";
-    private final TopicIdPartition topicIdPartition2 = new TopicIdPartition(topicName1, topicId1,
-        partition1);
+    private final TopicIdPartition topicIdPartition2 = new TopicIdPartition(topicId1,
+        partition1, topicName1);
+    private final TopicIdPartition topicIdPartitionWithNullTopic2 = new TopicIdPartition(topicId1,
+        new TopicPartition(null, partition1));
 
     @Test
     public void testEquals() {
         assertEquals(topicIdPartition0, topicIdPartition1);
         assertEquals(topicIdPartition1, topicIdPartition0);
+        assertEquals(topicIdPartitionWithNullTopic0, topicIdPartitionWithNullTopic1);
 
         assertNotEquals(topicIdPartition0, topicIdPartition2);
         assertNotEquals(topicIdPartition2, topicIdPartition0);
+        assertNotEquals(topicIdPartition0, topicIdPartitionWithNullTopic0);
+        assertNotEquals(topicIdPartitionWithNullTopic0, topicIdPartitionWithNullTopic2);
     }
 
     @Test
@@ -52,12 +62,20 @@ public void testHashCode() {
         assertEquals(Objects.hash(topicIdPartition0.topicId(), topicIdPartition0.topicPartition()),
             topicIdPartition0.hashCode());
         assertEquals(topicIdPartition0.hashCode(), topicIdPartition1.hashCode());
+
+        assertEquals(Objects.hash(topicIdPartitionWithNullTopic0.topicId(),
+            new TopicPartition(null, partition1)), topicIdPartitionWithNullTopic0.hashCode());
+        assertEquals(topicIdPartitionWithNullTopic0.hashCode(), topicIdPartitionWithNullTopic1.hashCode());
+
         assertNotEquals(topicIdPartition0.hashCode(), topicIdPartition2.hashCode());
+        assertNotEquals(topicIdPartition0.hashCode(), topicIdPartitionWithNullTopic0.hashCode());
+        assertNotEquals(topicIdPartitionWithNullTopic0.hashCode(), topicIdPartitionWithNullTopic2.hashCode());
     }
 
     @Test
     public void testToString() {
         assertEquals("vDiRhkpVQgmtSLnsAZx7lA:a_topic_name-1", topicIdPartition0.toString());
+        assertEquals("vDiRhkpVQgmtSLnsAZx7lA:null-1", topicIdPartitionWithNullTopic0.toString());
     }
 
 }