dt: Test for validate_only reponse code and semantics.

In `CreateTopicsResponseTest`.

(cherry picked from commit 0654662)

tests/rptest/clients/kcl.py

@@ -435,13 +435,13 @@ def create_topics(self,
         except:
             return []
 
-    def raw_create_topics(self, version, topics):
+    def raw_create_topics(self, version, topics, validate_only=False):
         assert version >= 0 and version <= 6, "version out of supported redpanda range for this API"
         create_topics_request = {
             'Version':
             version,
             'ValidateOnly':
-            False,
+            validate_only,
             'TimeoutMillis':
             60000,
             'Topics': [{

tests/rptest/tests/topic_creation_test.py

@@ -311,7 +311,8 @@ def __init__(self, test_context):
 
     # we don't really care about the name aside from its not being random
     # so just construct it from the partition count and replication factor
-    def create_topics(self, p_cnt, r_fac, n=1):
+
+    def create_topics(self, p_cnt, r_fac, n=1, validate_only=False):
         topics = []
         for i in range(0, n):
             topics.append({
@@ -320,7 +321,9 @@ def create_topics(self, p_cnt, r_fac, n=1):
                 'replication_factor': r_fac
             })
 
-        return self.kcl_client.create_topics(6, topics=topics)
+        return self.kcl_client.create_topics(6,
+                                             topics=topics,
+                                             validate_only=validate_only)
 
     def get_np(self, tp):
         return tp['NumPartitions']
@@ -390,6 +393,23 @@ def test_create_topic_response_configs(self):
             assert cleanup_policy[
                 'Source'] == self.DEFAULT_CONFIG_SOURCE, f"cleanup.policy = {cleanup_policy['Source']}, expected {self.DEFAULT_CONFIG_SOURCE}"
 
+    @cluster(num_nodes=3)
+    def test_create_topic_validate_only(self):
+        """
+        Validates that create topics calls with validate only flag return
+        the correct error code depending on whether or not the topic already
+        exists.
+        """
+
+        topic = self.create_topics(1, 1, validate_only=True)[0]
+        self.check_topic_resp(topic, 1, 1, self.SUCCESS_EC)
+
+        topic = self.create_topics(1, 1)[0]
+        self.check_topic_resp(topic, 1, 1, self.SUCCESS_EC)
+
+        topic = self.create_topics(1, 1, validate_only=True)[0]
+        self.check_topic_resp(topic, -1, -1, self.TOPIC_EXISTS_EC)
+
 
 class CreateSITopicsTest(RedpandaTest):
     def __init__(self, test_context):