non-json payloads on error responses are handled

nakadi-java-client/src/main/java/nakadi/Problem.java

@@ -145,6 +145,24 @@ public static Problem observerProblem(String title, String detail) {
         .type(OBSERVER_TYPE);
   }
 
+  /**
+   * Quick way to create a Problem object that indicates a server responded with an error but the message could not be converted to a Problem
+   * object automatically; typically this happens when a server responds with a non-json payload.
+   *
+   * @param statusCode the HTTP status code of the server's response
+   * @param body the HTTP body of the server's response
+   * @param errorDetail details about why it failed to convert the server's response to a Problem object
+   * @return a Problem object with a status of {@param statusCode} and a type of "about:contract_retryable"
+   */
+  public static Problem rawProblem(int statusCode, String body, String errorDetail) {
+    return new Problem()
+            .title(body)
+            .detail(errorDetail)
+            .data(SENTINEL_MAP)
+            .status(statusCode)
+            .type(CONTRACT_RETRYABLE_TYPE);
+  }
+
   public String toMessage() {
     if (message != null) {
       return message;

nakadi-java-client/src/main/java/nakadi/ProblemSupport.java

@@ -6,7 +6,7 @@ class ProblemSupport {
 
   static Problem toProblem(Response response, JsonSupport jsonSupport) {
     final String raw = response.responseBody().asString();
-    Problem problem = jsonSupport.fromJson(raw, Problem.class);
+    Problem problem = deserializeProblem(response.statusCode(), raw, jsonSupport);
 
     if (problem == null) {
       problem = Problem.noProblemo("no problem sent back from server", "", response.statusCode());
@@ -29,6 +29,14 @@ static Problem toProblem(Response response, JsonSupport jsonSupport) {
     return problem;
   }
 
+  static Problem deserializeProblem(int statusCode, String body, JsonSupport jsonSupport) {
+    try {
+      return jsonSupport.fromJson(body, Problem.class);
+    } catch(Exception e) {
+      return Problem.rawProblem(statusCode, body, e.getMessage());
+    }
+  }
+
   static <T> T throwProblem(int code, Problem problem, MetricCollector metricCollector) {
     if (code == 401) {
       metricCollector.mark(MetricCollector.Meter.http401);

nakadi-java-client/src/test/java/nakadi/ProblemSupportTest.java

@@ -5,6 +5,8 @@
 import java.io.Reader;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
@@ -30,6 +32,18 @@ public void toProblemForNakadi645() {
     assertEquals(401, problem.status());
   }
 
+  @Test
+  public void toProblemForNakadi357() {
+
+    // test fix for https://github.com/dehora/nakadi-java/issues/357
+
+    Response response = buildReponse("non-json-payload", 400);
+    Problem problem = ProblemSupport.toProblem(response, jsonSupport);
+    assertEquals("non-json-payload", problem.title());
+    assertEquals(400, problem.status());
+    assertEquals(Optional.of("java.lang.IllegalStateException: Expected BEGIN_OBJECT but was STRING at line 1 column 1 path $"), problem.detail());
+  }
+
   private Response buildReponse(String json, int code) {
     return new Response() {
       @Override public int statusCode() {