Runtime exceptions from chunkProccesor.process are not propagated

[rhajek]

Client code is not possible to handle runtime exceptions fired from chunk processing.

[majst01]

Can you give a example with Stacktrace ?

[rhajek]

I will prepare pull request, containing unit tests.

[rhajek]

Here is the example and unit test. This is the serious issue that causes the threads hangs/freeze because the client code is not notified when the async query is finished (onError/onComplete is not called).

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Query;

import org.junit.Assert;
import org.junit.Test;

public class ChunkQueryExceptionTest {
  private static Logger log = Logger.getLogger(ChunkQueryExceptionTest.class.getName());

  @Test
  public void testRuntimeException() throws Exception {

    InfluxDB client = InfluxDBFactory.connect("http://localhost:8086", "root", "root");

    CountDownLatch latch = new CountDownLatch(1);

    client.query(new Query("Select (*) from cpu", "telegraf"), 2,
        (cancellable, r) -> {
          log.info("processing result: " + r.toString());
          //any user code exception like parse 'null' to number...
          throw new java.lang.NumberFormatException("my-chunk-processing-error");
          // if any runtime exception is thrown while processing the chunk (processing si done in different thread)
          // - call.cancel() is not called, so client does not stop streaming
          // - client code is not notified that query was finished and hangs forever
          // - exception is swallowed, it is not possible to catch and handle it
          // - ugly workaround is to surround whole consumer lambda try/catch generic exception block

        }, () -> log.info("Completed."),

        (error) -> {
          // check that the runtime exceptions are propagated here
          if (error.getMessage().equals("my-chunk-processing-error")) {
            latch.countDown();
          }
          log.log(Level.SEVERE, "Exception occured while query", error);
        }
    );

    Assert.assertTrue(latch.await(5, TimeUnit.SECONDS));

  }

}

[fmachado]

Fixed by PR #640