influxdata / influxdata/influxdb-java

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

オープン
#688 コメント 7 件 リアクション 0 件 担当者 0 名 GitHub で見る

まだ誰も着手していません。

主要言語
Java
スター
1.2k
フォーク
469
PR マージ指標
30日以内にマージされた PR はありません

説明

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.

コントリビューションガイド

このリポジトリのコントリビューションガイドは索引されていません

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

調査の方向性

issue で説明されている BatchProcessor のキューへのエンキュー経路から始め、BatchOptions と遅延する OkHttp インターセプターを使用した、提供されている JUnit の例で動作を再現します。デフォルトのバッチサイズに関する動作、特に 2000 回目の書き込みを確認し、キューが満杯の間も非同期書き込みがブロックされなくなった時点で issue は完了したものとします。

索引モデルが issue の本文から書いたものです。

評価

技術スタック
java
領域
backend-api-design
issue の種類
バグ
難易度
3/5
見積もり時間
1〜2日
活発さ
停滞
明瞭さ
明確に書かれている
初心者へのやさしさ
35/100

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。