Merge pull request #19 from confluentinc/performance-test-error-handling

Check for errors during performance test by checking HTTP status.

src/main/java/io/confluent/kafkarest/tools/ConsumerPerformance.java

@@ -31,6 +31,7 @@
 import io.confluent.kafkarest.Versions;
 import io.confluent.kafkarest.entities.ConsumerInstanceConfig;
 import io.confluent.kafkarest.entities.CreateConsumerInstanceResponse;
+import io.confluent.rest.entities.ErrorMessage;
 
 public class ConsumerPerformance extends AbstractPerformanceTest {
 
@@ -41,6 +42,8 @@ public class ConsumerPerformance extends AbstractPerformanceTest {
   String deleteUrl;
   long consumedRecords = 0;
 
+  private final ObjectMapper jsonDeserializer = new ObjectMapper();
+
   public static void main(String[] args) throws Exception {
     if (args.length < 4) {
       System.out.println(
@@ -122,22 +125,27 @@ private <T> T request(String target, String method, byte[] entity, String entity
       if (entity != null) {
         connection.setRequestProperty("Content-Type", Versions.KAFKA_MOST_SPECIFIC_DEFAULT);
         connection.setRequestProperty("Content-Length", entityLength);
-        connection.setDoInput(true);
       }
-
+      connection.setDoInput(true);
       connection.setUseCaches(false);
-      if (method != "DELETE") {
-        connection.setDoOutput(true);
-      }
-
       if (entity != null) {
+        connection.setDoOutput(true);
         OutputStream os = connection.getOutputStream();
         os.write(entity);
         os.flush();
         os.close();
       }
 
-      if (method != "DELETE") {
+      int responseStatus = connection.getResponseCode();
+      if (responseStatus >= 400) {
+        InputStream es = connection.getErrorStream();
+        ErrorMessage errorMessage = jsonDeserializer.readValue(es, ErrorMessage.class);
+        es.close();
+        throw new RuntimeException(
+            String.format("Unexpected HTTP error status %d for %s request to %s: %s",
+                          responseStatus, method, target, errorMessage.getMessage()));
+      }
+      if (responseStatus != HttpURLConnection.HTTP_NO_CONTENT) {
         InputStream is = connection.getInputStream();
         T result = serializer.readValue(is, responseFormat);
         is.close();

src/main/java/io/confluent/kafkarest/tools/ProducerPerformance.java

@@ -28,6 +28,7 @@
 import io.confluent.kafkarest.Versions;
 import io.confluent.kafkarest.entities.TopicProduceRecord;
 import io.confluent.kafkarest.entities.TopicProduceRequest;
+import io.confluent.rest.entities.ErrorMessage;
 
 public class ProducerPerformance extends AbstractPerformanceTest {
 
@@ -41,6 +42,8 @@ public class ProducerPerformance extends AbstractPerformanceTest {
   byte[] requestEntity;
   byte[] buffer;
 
+  private final ObjectMapper jsonDeserializer = new ObjectMapper();
+
   public static void main(String[] args) throws Exception {
     if (args.length < 6) {
       System.out.println(
@@ -105,6 +108,15 @@ protected void doIteration(PerformanceStats.Callback cb) {
       os.flush();
       os.close();
 
+      int responseStatus = connection.getResponseCode();
+      if (responseStatus >= 400) {
+        InputStream es = connection.getErrorStream();
+        ErrorMessage errorMessage = jsonDeserializer.readValue(es, ErrorMessage.class);
+        es.close();
+        throw new RuntimeException(
+            String.format("Unexpected HTTP error status %d: %s",
+                          responseStatus, errorMessage.getMessage()));
+      }
       InputStream is = connection.getInputStream();
       while (is.read(buffer) > 0) {
         // Ignore output, just make sure we actually receive it