influxdata / influxdata/influxdb-java
The client act in a synchronous manner when BatchProcessorr#queue is full.
まだ誰も着手していません。
- 主要言語
- 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.
コントリビューションガイド
このリポジトリのコントリビューションガイドは索引されていません
はじめの一歩
- issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
- 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
- リポジトリをフォークし、ブランチを切って変更します。
- issue 番号を参照したプルリクエストを送ります。
調査の方向性
issue で説明されている BatchProcessor のキューへのエンキュー経路から始め、BatchOptions と遅延する OkHttp インターセプターを使用した、提供されている JUnit の例で動作を再現します。デフォルトのバッチサイズに関する動作、特に 2000 回目の書き込みを確認し、キューが満杯の間も非同期書き込みがブロックされなくなった時点で issue は完了したものとします。
索引モデルが issue の本文から書いたものです。
評価
- 技術スタック
- java
- 領域
- backend-api-design
- issue の種類
- バグ
- 難易度
- 3/5
- 見積もり時間
- 1〜2日
- 活発さ
- 停滞
- 明瞭さ
- 明確に書かれている
- 初心者へのやさしさ
- 35/100