influxdata / influxdata/influxdb-java

The client act in a synchronous manner when BatchProcessorr#queue is full.

Abierto
#688 7 comentarios 0 reacciones 0 asignados Ver en GitHub

Nadie ha tomado este issue todavía.

Lenguaje dominante
Java
Estrellas
1.2k
Forks
469
Métricas de merge de PR
Sin PR fusionados en 30 d

Descripción

I went through the `BatchProcessor` implementation and noticed that when the batch processor is lagging behind and is not able to clear the `queue`, **`InfluxDB.write(Point p)` becomes a blocking call even when one has configured it to write in an asynchronous manner.**
This is happening because the queue instance is of type LinkedBlockingQueue and `put(E e)` is used for enqueueing which blocks the thread trying to enqueue.
**IMO, this behaviour is very misleading, can block all the application threads who are trying to write whenever the write latencies on the influx-db server increases.**
I am using the `influxdb-java:2.15`
Below is the sample `Junit` test to simulate this.
```

import okhttp3.Interceptor;
import okhttp3.OkHttpClient;
import okhttp3.Response;
import org.influxdb.BatchOptions;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.junit.Test;

import java.io.IOException;

public class InfluxTest {

InfluxDB influxDB;

@Test
public void testConnection() {

connect();

for (int i=0;i<5000; ++i) {

long before = System.currentTimeMillis();
influxDB.write(Point.measurement("test").addField("a-field", 0).build());
long after = System.currentTimeMillis();
System.out.println(String.format("Time taken for writing %d times was %d", i, after-before));
}

}

private void connect() {

String influxDbConnectionURL = "http://localhost:8086";

influxDB = InfluxDBFactory.connect(influxDbConnectionURL, new OkHttpClient.Builder().addInterceptor(new Interceptor() {
@Override
public Response intercept(Chain chain) throws IOException {
try {
Thread.sleep(10000);
System.out.println("interacted with influx");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return chain.proceed(chain.request());
}
}));

influxDB.query(new Query("CREATE DATABASE " + '"' + "test" + '"'));
influxDB.setDatabase("test");

influxDB.enableBatch(BatchOptions.DEFAULTS.exceptionHandler(
(failedPoints, throwable) -> {
})
);

}

}

```
The default batch size is 1000, so the 2000th write call will be blocked for ~10000ms.

Guía de contribución

No hay ninguna guía de contribución indexada para este repositorio

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Línea de trabajo

Comience con la ruta de encolado de BatchProcessor descrita en el issue y reproduzca el comportamiento utilizando el ejemplo de JUnit proporcionado con BatchOptions y un interceptor de OkHttp con retraso. Verifique el comportamiento en torno al tamaño de batch predeterminado, especialmente en la escritura número 2000, y considere completado el issue cuando las escrituras asíncronas ya no se bloqueen mientras la cola esté llena.

Escrito por el modelo de indexación a partir del texto del issue.

Evaluación

Stack tecnológico
java
Área
backend-api-design
Tipo de issue
Error
Dificultad
3/5
Tiempo estimado
1-2 días
Estado de actividad
Estancado
Claridad
Bien especificado
Aptitud para principiantes
35/100

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.