fix issue #505 : Error messages from server not parsed correctly when using msgpack

src/main/java/org/influxdb/InfluxDBException.java

@@ -1,5 +1,12 @@
 package org.influxdb;
 
+import java.io.InputStream;
+
+import org.msgpack.core.MessagePack;
+import org.msgpack.core.MessageUnpacker;
+import org.msgpack.value.ImmutableMapValue;
+import org.msgpack.value.impl.ImmutableStringValueImpl;
+
 import com.squareup.moshi.JsonAdapter;
 import com.squareup.moshi.Moshi;
 
@@ -168,4 +175,20 @@ public static InfluxDBException buildExceptionForErrorState(final String errorBo
       return new InfluxDBException(errorBody);
     }
   }
+
+  /**
+   * Create corresponding InfluxDBException from the message pack error body.
+   * @param messagePackErrorBody
+   * @return
+   */
+  public static InfluxDBException buildExceptionForErrorState(final InputStream messagePackErrorBody) {
+    try {
+      MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(messagePackErrorBody);
+      ImmutableMapValue mapVal = (ImmutableMapValue) unpacker.unpackValue();
+      return InfluxDBException.buildExceptionFromErrorMessage(
+          mapVal.map().get(new ImmutableStringValueImpl("error")).toString());
+    } catch (Exception e) {
+      return new InfluxDBException(e);
+    }
+  }
 }

src/main/java/org/influxdb/impl/InfluxDBImpl.java

@@ -687,7 +687,11 @@ private <T> T execute(final Call<T> call) {
         return response.body();
       }
       try (ResponseBody errorBody = response.errorBody()) {
-        throw InfluxDBException.buildExceptionForErrorState(errorBody.string());
+        if (messagePack) {
+          throw InfluxDBException.buildExceptionForErrorState(errorBody.byteStream());
+        } else {
+          throw InfluxDBException.buildExceptionForErrorState(errorBody.string());
+        }
       }
     } catch (IOException e) {
       throw new InfluxDBIOException(e);

src/test/java/org/influxdb/BatchOptionsTest.java

@@ -365,28 +365,23 @@ public Thread newThread(Runnable r) {
   public void testHandlerOnRetryImpossible() throws InterruptedException {
 
     String dbName = "write_unittest_" + System.currentTimeMillis();
-    InfluxDB spy = spy(influxDB);
-    doThrow(DatabaseNotFoundException.class).when(spy).write(any(BatchPoints.class));
 
     try {
       BiConsumer<Iterable<Point>, Throwable> mockHandler = mock(BiConsumer.class);
       BatchOptions options = BatchOptions.DEFAULTS.exceptionHandler(mockHandler).flushDuration(100);
 
-      spy.createDatabase(dbName);
-      spy.setDatabase(dbName);
-      spy.enableBatch(options);
+      influxDB.setDatabase(dbName);
+      influxDB.enableBatch(options);
 
-      writeSomePoints(spy, 1);
+      writeSomePoints(influxDB, 1);
 
       Thread.sleep(200);
       verify(mockHandler, times(1)).accept(any(), any());
 
       QueryResult result = influxDB.query(new Query("select * from weather", dbName));
       Assertions.assertNull(result.getResults().get(0).getSeries());
-      Assertions.assertNull(result.getResults().get(0).getError());
     } finally {
-      spy.disableBatch();
-      spy.deleteDatabase(dbName);
+      influxDB.disableBatch();
     }
 
   }

src/test/java/org/influxdb/InfluxDBExceptionTest.java

@@ -0,0 +1,30 @@
+package org.influxdb;
+
+import org.influxdb.InfluxDBException.DatabaseNotFoundException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.platform.runner.JUnitPlatform;
+import org.junit.runner.RunWith;
+
+/**
+ * Test cases for InfluxDBException
+ *
+ * @author hoan.le [at] bonitoo.io
+ *
+ */
+
+@RunWith(JUnitPlatform.class)
+public class InfluxDBExceptionTest {
+
+  @Test
+  public void testBuildExceptionForMessagePackErrorState() {
+    DatabaseNotFoundException dbex = (DatabaseNotFoundException) InfluxDBException
+        .buildExceptionForErrorState(InfluxDBExceptionTest.class.getResourceAsStream("msgpack_errorBody.bin"));
+
+    Assertions.assertEquals("database not found: \"abc\"", dbex.getMessage());
+
+    InfluxDBException ex = InfluxDBException.buildExceptionForErrorState(InfluxDBExceptionTest.class.getResourceAsStream("invalid_msgpack_errorBody.bin"));
+    Assertions.assertTrue(ex.getCause() instanceof ClassCastException);
+
+  }
+}

src/test/resources/org/influxdb/invalid_msgpack_errorBody.bin

@@ -0,0 +1 @@
+e not found: "abc"

src/test/resources/org/influxdb/msgpack_errorBody.bin

@@ -0,0 +1 @@
+��error�database not found: "abc"