You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "masfworld (via GitHub)" <gi...@apache.org> on 2025/08/22 15:13:51 UTC
[I] [Bug]: IcebergIO- RecordWriterManager fails with "Expected all data writers to be closed" when exceptions occur during write [beam]
masfworld opened a new issue, #35940:
URL: https://github.com/apache/beam/issues/35940
### What happened?
### Description
When writing to Iceberg tables with Managed.write(Managed.ICEBERG) in a streaming pipeline, I consistently see the following error in Dataflow workers:
```
java.lang.IllegalArgumentException: Expected all data writers to be closed, but found 1 data writer(s) still open
at org.apache.beam.sdk.io.iceberg.RecordWriterManager.close(RecordWriterManager.java:395)
at org.apache.beam.sdk.io.iceberg.WriteGroupedRowsToFiles$WriteGroupedRowsToFilesDoFn.processElement(WriteGroupedRowsToFiles.java:110)
```
Once it starts, this error repeats many times after 20-30 min, but not always in the same moment; the metric activeIcebergWriters rises (e.g., 50+) and the job stops writing new data. There is no consistent user-code exception preceding it. It occurs even after we serialize the sink to a single lane and reduce parallelism to the minimum.
### Steps to reproduce
Minimal pipeline (simplified for reproducibility):
```
public class MainSimpleCount {
private static final Logger LOG = LoggerFactory.getLogger(MainSimpleCount.class);
public static void main(String[] args) throws Exception {
IcebergPipelineOptions options = PipelineOptionsFactory.fromArgs(args)
.withValidation()
.as(IcebergPipelineOptions.class);
options.setStreaming(true);
options.setEnableStreamingEngine(true);
Pipeline pipeline = Pipeline.create(options);
// Parse JSON into Rows (note: schema fields are UPPERCASE per your error log)
PCollection<Row> input = pipeline
.apply("ReadFromPubSub", PubsubIO.readStrings().fromSubscription(options.getInputSubscription()))
.apply("ParseJsonToRow", ParDo.of(new ParseJsonToRow()))
.setRowSchema(Schemas.INPUT_SCHEMA);
// Simple 1-minute fixed window on processing time (no triggers, no accumulation)
PCollection<Row> windowed = input.apply("Window1m",
Window.<Row>into(FixedWindows.of(Duration.standardMinutes(1))));
// Key by ID we derive from UPPERCASE fields: a + '|' + b
PCollection<KV<String, Row>> keyed = windowed.apply("KeyById",
WithKeys.of((Row r) -> {
String a = r.getString("a");
String b = r.getString("b");
return a + "|" + b;
}))
.setCoder(org.apache.beam.sdk.coders.KvCoder.of(
org.apache.beam.sdk.coders.StringUtf8Coder.of(),
org.apache.beam.sdk.schemas.SchemaCoder.of(Schemas.INPUT_SCHEMA)));
// Group by ID
PCollection<KV<String, Iterable<Row>>> grouped =
keyed.apply("GroupById", GroupByKey.create());
// Count per key -> build output Row(ID, count_id)
PCollection<Row> counted = grouped.apply("CountRows",
ParDo.of(new DoFn<KV<String, Iterable<Row>>, Row>() {
@ProcessElement
public void processElement(ProcessContext c) {
String Id = c.element().getKey();
long count = 0;
for (Row r : c.element().getValue()) {
count++;
}
Row outRow = Row.withSchema(Schemas.CC_COUNT_SCHEMA)
.addValues(Id, count)
.build();
c.output(outRow);
}
}))
.setRowSchema(Schemas.CC_COUNT_SCHEMA);
// Write to Iceberg
counted.apply("WriteToIceberg", Managed.write(Managed.ICEBERG)
.withConfig(ImmutableMap.<String, Object>builder()
.put("table", "test_namespace.prices_iceberg_table_1m_id_count")
.put("catalog_name", "snowflake_open_catalog_internal")
.put("catalog_properties", ImmutableMap.<String, String>builder()
.put("type", "rest")
.put("uri", "https://XXXX.XXXXX.gcp.snowflakecomputing.com/polaris/api/catalog/")
.put("oauth2-server-uri", "https://XXXX.XXXX.gcp.snowflakecomputing.com/polaris/api/catalog/v1/oauth/tokens")
.put("header.X-Iceberg-Access-Delegation", "vended-credentials")
.put("credential", options.getCredential())
.put("warehouse", "snowflake_open_catalog_internal")
.put("scope", "PRINCIPAL_ROLE:principal_role")
.put("io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO")
.put("token-refresh-enabled", "false")
.build())
// only keep the two table columns
.put("keep", java.util.List.of("id", "count_id"))
.put("triggering_frequency_seconds", 120)
.build()));
pipeline.run();
}
}
```
Environment:
- Apache Beam: 2.67.0
- Runner: Google Cloud Dataflow (streaming)
- Input source: Pub/Sub → JSON → Beam Row
- Snowflake Open Catalog for Iceberg catalog
### Issue Priority
Priority: 1 (data loss / total loss of function)
### Issue Components
- [ ] Component: Python SDK
- [x] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Samza Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [x] Component: Google Cloud Dataflow Runner
--
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-unsubscribe@beam.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: IcebergIO- RecordWriterManager fails with "Expected all data writers to be closed" when exceptions occur during write [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on issue #35940:
URL: https://github.com/apache/beam/issues/35940#issuecomment-3228694984
Thanks for checking back in @masfworld. My pipeline was using HadoopCatalog as well.
I'll comment in the new tracker you created
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: IcebergIO- RecordWriterManager fails with "Expected all data writers to be closed" when exceptions occur during write [beam]
Posted by "masfworld (via GitHub)" <gi...@apache.org>.
masfworld commented on issue #35940:
URL: https://github.com/apache/beam/issues/35940#issuecomment-3223026002
@ahmedabu98 Let me close this issue because the job works fine using Hadoop Iceberg catalog. I think the error is using an external catalog with GCS. There is a bug in Iceberg `v1.6.x` where credentials are not refreshed. It has been fixed in `v1.7.0`.
https://github.com/apache/iceberg/pull/11282
I think it's very important to update iceberg dependencies in Apache Beam
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: IcebergIO- RecordWriterManager fails with "Expected all data writers to be closed" when exceptions occur during write [beam]
Posted by "masfworld (via GitHub)" <gi...@apache.org>.
masfworld closed issue #35940: [Bug]: IcebergIO- RecordWriterManager fails with "Expected all data writers to be closed" when exceptions occur during write
URL: https://github.com/apache/beam/issues/35940
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: IcebergIO- RecordWriterManager fails with "Expected all data writers to be closed" when exceptions occur during write [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on issue #35940:
URL: https://github.com/apache/beam/issues/35940#issuecomment-3217017165
Hey @masfworld, I've been running a streaming pipeline for over an hour now, doing the same Pubsub -> JsonToRow -> Iceberg conversion and have not encountered that error. I'll keep it running and check again later, but I think it would be best to open a Google support ticket so we can take a look at your job
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug]: IcebergIO- RecordWriterManager fails with "Expected all data writers to be closed" when exceptions occur during write [beam]
Posted by "masfworld (via GitHub)" <gi...@apache.org>.
masfworld commented on issue #35940:
URL: https://github.com/apache/beam/issues/35940#issuecomment-3219144356
Maybe the problem is the configuration with Snowflake open Catalog... 🤔. Let me check
--
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-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org