influxdata / influxdata/influxdb-java
The client act in a synchronous manner when BatchProcessorr#queue is full.
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
- Lee el issue completo y luego la guía de contribución del proyecto.
- Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
- Haz un fork del repositorio y trabaja en una rama.
- 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