diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
index 8f73c8822adcc..d101f440720b9 100644
--- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java
@@ -591,6 +591,46 @@ public static String formatBytes(long bytes) {
         }
     }
 
+    /**
+     * Create a string representation of an array joined by the given separator
+     * @param strs The array of items
+     * @param separator The separator
+     * @return The string representation.
+     */
+    public static <T> String join(T[] strs, String separator) {
+        return join(Arrays.asList(strs), separator);
+    }
+
+    /**
+     * Create a string representation of a collection joined by the given separator
+     * @param collection The list of items
+     * @param separator The separator
+     * @return The string representation.
+     */
+    public static <T> String join(Collection<T> collection, String separator) {
+        Objects.requireNonNull(collection);
+        return mkString(collection.stream(), "", "", separator);
+    }
+
+    /**
+     * Create a string representation of a stream surrounded by `begin` and `end` and joined by `separator`.
+     *
+     * @return The string representation.
+     */
+    public static <T> String mkString(Stream<T> stream, String begin, String end, String separator) {
+        Objects.requireNonNull(stream);
+        StringBuilder sb = new StringBuilder();
+        sb.append(begin);
+        Iterator<T> iter = stream.iterator();
+        while (iter.hasNext()) {
+            sb.append(iter.next());
+            if (iter.hasNext())
+                sb.append(separator);
+        }
+        sb.append(end);
+        return sb.toString();
+    }
+
     /**
      *  Converts a {@code Map} class into a string, concatenating keys and values
      *  Example:
diff --git a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
index bd38ccabd900d..dfdb8834c1e17 100755
--- a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java
@@ -1043,4 +1043,18 @@ public int hashCode() {
             return key.hashCode();
         }
     }
+
+    @Test
+    public void testJoin() {
+        assertEquals("", Utils.join(Collections.emptyList(), ","));
+        assertEquals("1", Utils.join(asList("1"), ","));
+        assertEquals("1,2,3", Utils.join(asList(1, 2, 3), ","));
+    }
+
+    @Test
+    public void testMkString() {
+        assertEquals("[]", Utils.mkString(Stream.empty(), "[", "]", ","));
+        assertEquals("(1)", Utils.mkString(Stream.of("1"), "(", ")", ","));
+        assertEquals("{1,2,3}", Utils.mkString(Stream.of(1, 2, 3), "{", "}", ","));
+    }
 }