[DCOS-59064] Improve status check to verify topic successfully create…

…d in ZK (#458)

* Added conditional status to check failure at topic creation

* Changed condition to check for successful topic creation

The condition now checks for "Created topic" sub-string as a part of response message to provide appropriate status response

* Updated Response code to server-side specific

frameworks/kafka/src/main/java/com/mesosphere/sdk/kafka/api/TopicResource.java

@@ -60,7 +60,16 @@ public Response createTopic(
         try {
             int partCount = Integer.parseInt(partitionCount);
             int replFactor = Integer.parseInt(replicationFactor);
-            return ResponseUtils.jsonOkResponse(cmdExecutor.createTopic(name, partCount, replFactor));
+            JSONObject result = cmdExecutor.createTopic(name, partCount, replFactor);
+            String message = result.getString("message");
+            Response.Status status;
+            if (message.contains("Created topic")) {
+                status = Response.Status.OK;
+            } else {
+                status = Response.Status.INTERNAL_SERVER_ERROR;
+				log.error("Failed to verify the topic creation: " + name + " with output: " + message);
+            }
+            return ResponseUtils.jsonResponse(result, status);
         } catch (Exception ex) {
             log.error("Failed to create topic: " + name + " with exception: " + ex);
             return Response.serverError().build();