leonardBang commented on code in PR #12:
URL:
https://github.com/apache/flink-connector-mongodb/pull/12#discussion_r1244595332
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/sink/writer/MongoWriter.java:
##########
@@ -105,10 +115,35 @@ public MongoWriter(
// Initialize the mongo client.
this.mongoClient = MongoClients.create(connectionOptions.getUri());
+
+ if (!flushOnlyOnCheckpoint() && writeOptions.getBatchIntervalMs() > 0)
{
+ this.scheduler =
+ Executors.newScheduledThreadPool(1, new
ExecutorThreadFactory("mongo-writer"));
+
+ this.scheduledFuture =
+ this.scheduler.scheduleWithFixedDelay(
+ () -> {
+ synchronized (MongoWriter.this) {
+ if (!closed &&
isOverMaxBatchIntervalLimit()) {
+ try {
+ doBulkWrite();
+ } catch (Exception e) {
+ flushException = e;
+ }
+ }
+ }
+ },
+ writeOptions.getBatchIntervalMs(),
+ writeOptions.getBatchIntervalMs(),
Review Comment:
The `batchIntervalMs` could be initialized earlier during construct the
MongoDBWriter
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/sink/writer/MongoWriter.java:
##########
@@ -181,4 +238,14 @@ private boolean isOverMaxBatchIntervalLimit() {
long lastSentInterval = System.currentTimeMillis() - lastSendTime;
return bulkFlushInterval != -1 && lastSentInterval >=
bulkFlushInterval;
}
+
+ private boolean flushOnlyOnCheckpoint() {
+ return writeOptions.getBatchIntervalMs() == -1 &&
writeOptions.getBatchSize() == -1;
Review Comment:
We can optimize to calculate once in constructor
##########
flink-connector-mongodb/src/main/java/org/apache/flink/connector/mongodb/sink/writer/MongoWriter.java:
##########
@@ -131,8 +168,28 @@ public void flush(boolean endOfInput) throws IOException {
}
@Override
- public void close() {
- mongoClient.close();
+ public synchronized void close() throws Exception {
+ if (!closed) {
+ if (scheduledFuture != null) {
+ scheduledFuture.cancel(false);
+ scheduler.shutdown();
+ }
+
+ if (!bulkRequests.isEmpty()) {
+ try {
+ doBulkWrite();
+ } catch (Exception e) {
+ LOG.warn("Writing records to MongoDB failed when closing
MongoWriter", e);
Review Comment:
error level?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
