ahmedabu98 commented on code in PR #32102:
URL: https://github.com/apache/beam/pull/32102#discussion_r1716250544
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/WriteGroupedRowsToFiles.java:
##########
@@ -75,28 +84,35 @@ private org.apache.iceberg.catalog.Catalog getCatalog() {
return catalog;
}
- private RecordWriter createWriter(IcebergDestination destination) throws
IOException {
- return new RecordWriter(getCatalog(), destination, "-" +
UUID.randomUUID());
- }
-
@ProcessElement
public void processElement(
- ProcessContext c, @Element KV<ShardedKey<Row>, Iterable<Row>> element)
throws Exception {
+ ProcessContext c,
+ @Element KV<ShardedKey<Row>, Iterable<Row>> element,
+ BoundedWindow window,
+ PaneInfo pane)
+ throws Exception {
Row destMetadata = element.getKey().getKey();
IcebergDestination destination =
dynamicDestinations.instantiateDestination(destMetadata);
- RecordWriter writer = createWriter(destination);
+ WindowedValue<IcebergDestination> windowedDestination =
+ WindowedValue.of(destination, window.maxTimestamp(), window, pane);
+ RecordWriterManager writer =
+ new RecordWriterManager(getCatalog(), filePrefix, maxFileSize,
Integer.MAX_VALUE);
Review Comment:
Done
--
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: github-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
