influxdata / influxdata/influxdb-java
The client act in a synchronous manner when BatchProcessorr#queue is full.
Nessuno ha ancora preso questa issue.
- Lingua principale
- Java
- Stelle
- 1.2k
- Fork
- 469
- Metriche di merge delle PR
- Nessuna PR unita negli ultimi 30g
Descrizione
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.
Guida per i contributori
Nessuna guida per i contributori indicizzata per questo repository
Come iniziare
- Leggi tutta la issue e poi la guida ai contributi del progetto.
- Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
- Fai un fork del repository e lavora su un branch.
- Apri una pull request che faccia riferimento al numero della issue.
Direzione di ricerca
Iniziare con il percorso di accodamento di BatchProcessor descritto nell’issue e riprodurre il comportamento utilizzando l’esempio JUnit fornito con BatchOptions e un interceptor OkHttp ritardato. Verificare il comportamento relativo alla dimensione batch predefinita, in particolare alla 2000ª scrittura, e considerare l’issue completata quando le scritture asincrone non si bloccano più mentre la coda è piena.
Scritto dal modello di indicizzazione a partire dal testo della issue.
Valutazione
- Stack tecnologico
- java
- Ambito
- backend-api-design
- Tipo di issue
- Bug
- Difficoltà
- 3/5
- Tempo stimato
- 1-2 giorni
- Stato di attività
- Ferma
- Chiarezza
- Specificata chiaramente
- Idoneità per principianti
- 35/100