[FLINK-32446][Connectors/MongoDB] MongoWriter should regularly check whether the last write time is over limit #12
Conversation
…whether the last write time is over limit
|
Hi @leonardBang, |
leonardBang
left a comment
leonardBang
left a comment
There was a problem hiding this comment.
Thanks @Jiabao-Sun for the contribution, I left minor comments
| writeOptions.getBatchIntervalMs(), | ||
| writeOptions.getBatchIntervalMs(), |
There was a problem hiding this comment.
The batchIntervalMs could be initialized earlier during construct the MongoDBWriter
| try { | ||
| doBulkWrite(); | ||
| } catch (Exception e) { | ||
| LOG.warn("Writing records to MongoDB failed when closing MongoWriter", e); |
| } | ||
|
|
||
| private boolean flushOnlyOnCheckpoint() { | ||
| return writeOptions.getBatchIntervalMs() == -1 && writeOptions.getBatchSize() == -1; |
There was a problem hiding this comment.
We can optimize to calculate once in constructor
leonardBang
left a comment
leonardBang
left a comment
There was a problem hiding this comment.
Thanks @Jiabao-Sun for the update, LGTM, let's wait the CI green
Thanks @leonardBang, the CI seems passed. |
|
|
||
| @Override | ||
| public void write(IN element, Context context) throws IOException, InterruptedException { | ||
| public synchronized void write(IN element, Context context) |
There was a problem hiding this comment.
@Jiabao-Sun just curious, wouldn't adding synchronized here hurt the performance for write operation?
There was a problem hiding this comment.
Thanks @6harat with this comment.
Adding synchronized here because we may add an other thread to check whether the last write time is over limit.
The thread may call doBulkWrite method and clear the bulkRequests list which is write by write method introduces a race condition.
…whether the latest flush time is arriving This closes #12.
3 participants
Mongo sink waits for new record to write previous records. I have a upsert-kafka topic filled that has already some events. I start a new upsert-kafka to mongo db sink job. I expect all the data from the topic to be loaded to mongodb right away. But instead, only the first record is written to mongo db. The rest of the records don’t arrive in mongodb until a new event is written to kafka topic. The new event that was written is delayed until the next event arrives.
To prevent this problem, the MongoWriter should regularly check whether the last write time is over the specific limit.