bobby-richard commented on code in PR #10045:
URL: https://github.com/apache/pinot/pull/10045#discussion_r1063463950
##########
pinot-connectors/pinot-flink-connector/src/main/java/org/apache/pinot/connector/flink/sink/PinotSinkFunction.java:
##########
@@ -117,31 +123,22 @@ public void close()
@Override
public void invoke(T value, Context context)
throws Exception {
- _segmentWriter.collect(_recordConverter.convertToRow(value));
+ GenericRow row = _recordConverter.convertToRow(value);
+ _pendingRows.add(row);
+ _segmentWriter.collect(row);
_segmentNumRecord++;
- if (_segmentNumRecord > _segmentFlushMaxNumRecords) {
+ if (_segmentNumRecord >= _segmentFlushMaxNumRecords) {
flush();
}
}
- @Override
- public void snapshotState(FunctionSnapshotContext functionSnapshotContext)
- throws Exception {
- throw new UnsupportedOperationException("snapshotState is invoked in Pinot
sink");
- }
-
- @Override
- public void initializeState(FunctionInitializationContext
functionInitializationContext)
- throws Exception {
- // no initialization needed
- // ...
- }
-
private void flush()
throws Exception {
URI segmentURI = _segmentWriter.flush();
LOG.info("Pinot segment writer flushed with {} records to {}",
_segmentNumRecord, segmentURI);
_segmentNumRecord = 0;
+ _pendingRows.clear();
Review Comment:
I tried this at first, but upload appears to happen asynchronously and the
segmentWriter has already been cleared before the upload.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
