jack-moseley commented on a change in pull request #3466:
URL: https://github.com/apache/gobblin/pull/3466#discussion_r805090882
##########
File path:
gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java
##########
@@ -368,29 +369,40 @@ private void flush(String dbName, String tableName)
throws IOException {
* Note that this is one of the place where the materialization of
aggregated metadata happens.
* When there's a change of {@link OperationType}, it also interrupts
metadata aggregation,
* and triggers materialization of metadata.
+ *
+ * @param emitAllErrors if true, go through all datasets in {@link
#datasetErrorMap} and emit GTE for each. We only want
+ * to do this if a watermark is is being advanced.
+ *
* @throws IOException
*/
- @Override
- public void flush() throws IOException {
+ public void flush(boolean emitAllErrors) throws IOException {
log.info(String.format("start to flushing %s records",
String.valueOf(recordCount.get())));
for (String tableString : tableOperationTypeMap.keySet()) {
List<String> tid =
Splitter.on(TABLE_NAME_DELIMITER).splitToList(tableString);
flush(tid.get(0), tid.get(1));
}
tableOperationTypeMap.clear();
recordCount.lazySet(0L);
- // Emit events for all current errors, since the GMCE watermark will be
advanced
- for (Map.Entry<String, Map<String, GobblinMetadataException>> entry :
datasetErrorMap.entrySet()) {
- for (GobblinMetadataException exception : entry.getValue().values()) {
- submitFailureEvent(exception);
+ if (emitAllErrors) {
Review comment:
As discussed removed the flush from close() instead.
--
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]
