influxdata / influxdata/influxdb-java

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

Offen
#688 7 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen

Dieses Issue hat noch niemand übernommen.

Vorherrschende Sprache
Java
Sterne
1.2k
Forks
469
PR-Merge-Kennzahlen
Keine gemergten PRs in 30 T.

Beschreibung

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.

Beitragsleitfaden

Für dieses Repository ist kein Beitragsleitfaden indexiert

Erste Schritte

  1. Lies das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreib ins Issue, dass du es übernimmst — das erspart doppelte Arbeit.
  3. Forke das Repository und arbeite in einem Branch.
  4. Öffne einen Pull Request, der die Issue-Nummer nennt.

Rechercherichtung

Beginnen Sie mit dem in der Issue beschriebenen Pfad zum Einreihen in die BatchProcessor-Warteschlange und reproduzieren Sie das Verhalten anhand des bereitgestellten JUnit-Beispiels mit BatchOptions und einem verzögerten OkHttp-Interceptor. Überprüfen Sie das Verhalten rund um die standardmäßige Batch-Größe, insbesondere beim 2000. Schreibvorgang, und betrachten Sie die Issue als abgeschlossen, sobald asynchrone Schreibvorgänge nicht mehr blockieren, während die Warteschlange voll ist.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Bewertung

Tech-Stack
java
Bereich
backend-api-design
Issue-Typ
Bug
Schwierigkeit
3/5
Geschätzter Aufwand
1-2 Tage
Aktivitätsstatus
Veraltet
Klarheit
Klar beschrieben
Anfängerfreundlichkeit
35/100

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.