This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 6745a98 [INLONG-2174] Make the flush operation synchronized in
clickhouse sink to prevent data loss (#2175)
6745a98 is described below
commit 6745a98a901ded7b5427132bf4817ac16a1ef4f7
Author: TianqiWan <[email protected]>
AuthorDate: Mon Jan 17 19:54:16 2022 +0800
[INLONG-2174] Make the flush operation synchronized in clickhouse sink to
prevent data loss (#2175)
Co-authored-by: tianqiwan <[email protected]>
---
.../executor/ClickHouseAppendExecutor.java | 75 +++++++---------------
.../executor/ClickHouseUpsertExecutor.java | 61 +++---------------
2 files changed, 34 insertions(+), 102 deletions(-)
diff --git
a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/clickhouse/executor/ClickHouseAppendExecutor.java
b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/clickhouse/executor/ClickHouseAppendExecutor.java
index 19c5b22..d5593df 100644
---
a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/clickhouse/executor/ClickHouseAppendExecutor.java
+++
b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/clickhouse/executor/ClickHouseAppendExecutor.java
@@ -22,7 +22,6 @@ import org.apache.inlong.sort.formats.common.FormatInfo;
import org.apache.inlong.sort.flink.clickhouse.ClickHouseRowConverter;
import org.apache.inlong.sort.protocol.sink.ClickHouseSinkInfo;
import org.apache.flink.api.java.tuple.Tuple2;
-import
org.apache.flink.shaded.curator.org.apache.curator.shaded.com.google.common.util.concurrent.AbstractExecutionThreadService;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,23 +43,18 @@ public class ClickHouseAppendExecutor implements
ClickHouseExecutor {
private final FormatInfo[] formatInfos;
- private final int flushIntervalSecond;
-
private final int maxRetries;
private transient ClickHousePreparedStatement stmt;
private transient List<Row> batch = new ArrayList<>();
- private transient ExecuteBatchService executeBatchService;
-
public ClickHouseAppendExecutor(
String insertSql,
FormatInfo[] formatInfos,
ClickHouseSinkInfo clickHouseSinkInfo) {
this.insertSql = insertSql;
this.formatInfos = formatInfos;
- this.flushIntervalSecond = clickHouseSinkInfo.getFlushInterval();
this.maxRetries = clickHouseSinkInfo.getWriteMaxRetryTimes();
}
@@ -70,8 +64,6 @@ public class ClickHouseAppendExecutor implements
ClickHouseExecutor {
batch = new ArrayList<>();
}
stmt = (ClickHousePreparedStatement)
connection.prepareStatement(insertSql);
- executeBatchService = new ExecuteBatchService();
- executeBatchService.startAsync();
}
@Override
@@ -81,63 +73,44 @@ public class ClickHouseAppendExecutor implements
ClickHouseExecutor {
@Override
public synchronized void executeBatch() throws IOException {
- if (executeBatchService.isRunning()) {
- this.notifyAll();
- } else {
- throw new IOException("executor unexpectedly terminated",
executeBatchService.failureCause());
+ try {
+ if (!batch.isEmpty()) {
+ for (Row row : batch) {
+ ClickHouseRowConverter.setRow(stmt, formatInfos, row);
+ stmt.addBatch();
+ }
+ attemptExecuteBatch();
+ }
+ } catch (Exception exception) {
+ throw new IOException("Flush data to clickhouse failed! " +
exception);
}
}
@Override
public void closeStatement() throws SQLException {
- if (executeBatchService != null) {
- executeBatchService.stopAsync().awaitTerminated();
- }
-
if (stmt != null) {
stmt.close();
stmt = null;
}
}
- private class ExecuteBatchService extends AbstractExecutionThreadService {
-
- private ExecuteBatchService() {
- }
-
- protected void run() throws Exception {
- while (isRunning()) {
- synchronized (ClickHouseAppendExecutor.this) {
- ClickHouseAppendExecutor.this.wait(flushIntervalSecond *
1000L);
- if (!batch.isEmpty()) {
- for (Row row : batch) {
- ClickHouseRowConverter.setRow(stmt, formatInfos,
row);
- stmt.addBatch();
- }
- attemptExecuteBatch();
- }
+ private void attemptExecuteBatch() throws IOException {
+ for (int i = 1; i <= maxRetries; i++) {
+ try {
+ stmt.executeBatch();
+ batch.clear();
+ break;
+ } catch (SQLException e) {
+ LOG.error("ClickHouse executeBatch error, retry times = {}",
i, e);
+ if (i >= maxRetries) {
+ throw new IOException(e);
}
- }
- }
- private void attemptExecuteBatch() throws IOException {
- for (int i = 1; i <= maxRetries; i++) {
try {
- stmt.executeBatch();
- batch.clear();
- break;
- } catch (SQLException e) {
- LOG.error("ClickHouse executeBatch error, retry times =
{}", i, e);
- if (i >= maxRetries) {
- throw new IOException(e);
- }
-
- try {
- Thread.sleep((1000L * i));
- } catch (InterruptedException ex) {
- Thread.currentThread().interrupt();
- throw new IOException("unable to flush; interrupted
while doing another attempt", e);
- }
+ Thread.sleep((1000L * i));
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw new IOException("unable to flush; interrupted while
doing another attempt", e);
}
}
}
diff --git
a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/clickhouse/executor/ClickHouseUpsertExecutor.java
b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/clickhouse/executor/ClickHouseUpsertExecutor.java
index b85e959..9d7a3ba 100644
---
a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/clickhouse/executor/ClickHouseUpsertExecutor.java
+++
b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/clickhouse/executor/ClickHouseUpsertExecutor.java
@@ -22,7 +22,6 @@ import org.apache.inlong.sort.formats.common.FormatInfo;
import org.apache.inlong.sort.flink.clickhouse.ClickHouseRowConverter;
import org.apache.inlong.sort.protocol.sink.ClickHouseSinkInfo;
import org.apache.flink.api.java.tuple.Tuple2;
-import
org.apache.flink.shaded.curator.org.apache.curator.shaded.com.google.common.util.concurrent.AbstractExecutionThreadService;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,8 +49,6 @@ public class ClickHouseUpsertExecutor implements
ClickHouseExecutor {
private final FormatInfo[] formatInfos;
- private final int flushIntervalSecond;
-
private final int maxRetries;
private final List<Row> insertBatch = new ArrayList<>();
@@ -64,8 +61,6 @@ public class ClickHouseUpsertExecutor implements
ClickHouseExecutor {
private transient ClickHousePreparedStatement deleteStmt;
- private transient ExecuteBatchService executeBatchService;
-
public ClickHouseUpsertExecutor(
String insertSql,
String updateSql,
@@ -76,7 +71,6 @@ public class ClickHouseUpsertExecutor implements
ClickHouseExecutor {
this.updateSql = updateSql;
this.deleteSql = deleteSql;
this.formatInfos = formatInfos;
- this.flushIntervalSecond = clickHouseSinkInfo.getFlushInterval();
this.maxRetries = clickHouseSinkInfo.getWriteMaxRetryTimes();
}
@@ -85,8 +79,6 @@ public class ClickHouseUpsertExecutor implements
ClickHouseExecutor {
insertStmt = (ClickHousePreparedStatement)
connection.prepareStatement(insertSql);
updateStmt = (ClickHousePreparedStatement)
connection.prepareStatement(updateSql);
deleteStmt = (ClickHousePreparedStatement)
connection.prepareStatement(deleteSql);
- executeBatchService = new ExecuteBatchService();
- executeBatchService.startAsync();
}
@Override
@@ -101,19 +93,16 @@ public class ClickHouseUpsertExecutor implements
ClickHouseExecutor {
@Override
public synchronized void executeBatch() throws IOException {
- if (executeBatchService.isRunning()) {
- notifyAll();
- } else {
- throw new IOException("executor unexpectedly terminated",
executeBatchService.failureCause());
+ try {
+ processBatch(insertStmt, insertBatch);
+ processBatch(deleteStmt, deleteBatch);
+ } catch (Exception exception) {
+ throw new IOException("Flush data to clickhouse failed! " +
exception);
}
}
@Override
public void closeStatement() throws SQLException {
- if (executeBatchService != null) {
- executeBatchService.stopAsync().awaitTerminated();
- }
-
for (ClickHouseStatement stmt : Arrays.asList(insertStmt, updateStmt,
deleteStmt)) {
if (stmt != null) {
stmt.close();
@@ -121,43 +110,13 @@ public class ClickHouseUpsertExecutor implements
ClickHouseExecutor {
}
}
- private class ExecuteBatchService extends AbstractExecutionThreadService {
-
- private ExecuteBatchService() {
- }
-
- protected void run() throws Exception {
- while (isRunning()) {
- synchronized (ClickHouseUpsertExecutor.this) {
- wait(flushIntervalSecond * 1000L);
- processInsertBatch(insertStmt, insertBatch);
- processDeleteBatch(deleteStmt, deleteBatch);
- }
- }
- }
-
- private void processDeleteBatch(ClickHousePreparedStatement stmt,
List<Row> batch)
- throws SQLException {
- if (!batch.isEmpty()) {
- for (Row row : batch) {
- ClickHouseRowConverter.setRow(stmt, formatInfos, row);
- stmt.executeUpdate();
- }
+ private void processBatch(ClickHousePreparedStatement stmt, List<Row>
batch) throws Exception {
+ if (!batch.isEmpty()) {
+ for (Row row : batch) {
+ ClickHouseRowConverter.setRow(stmt, formatInfos, row);
+ stmt.addBatch();
}
- }
-
- private void processInsertBatch(ClickHousePreparedStatement stmt,
List<Row> batch)
- throws SQLException, IOException {
- if (!batch.isEmpty()) {
- for (Row row : batch) {
- ClickHouseRowConverter.setRow(stmt, formatInfos, row);
- stmt.addBatch();
- }
- attemptExecuteBatch(stmt, batch);
- }
- }
- private void attemptExecuteBatch(ClickHousePreparedStatement stmt,
List<Row> batch) throws IOException {
for (int i = 1; i <= maxRetries; i++) {
try {
stmt.executeBatch();
