You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "mutianf (via GitHub)" <gi...@apache.org> on 2023/11/28 21:42:50 UTC
[PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
mutianf opened a new pull request, #29548:
URL: https://github.com/apache/beam/pull/29548
Batcher#closeAsync already sends out remaining elements and wait on them to finish before closing. Calling closeAsync let us set a timeout on the future and it should not take longer than the operation timeout.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "igorbernstein2 (via GitHub)" <gi...@apache.org>.
igorbernstein2 commented on code in PR #29548:
URL: https://github.com/apache/beam/pull/29548#discussion_r1414625981
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableServiceImpl.java:
##########
@@ -470,50 +477,50 @@ static class BigtableWriterImpl implements Writer {
private String projectId;
private String instanceId;
private String tableId;
+ private Duration closeWaitTimeout;
private Distribution bulkSize = Metrics.distribution("BigTable-" + tableId, "batchSize");
private Distribution latency = Metrics.distribution("BigTable-" + tableId, "batchLatencyMs");
BigtableWriterImpl(
- BigtableDataClient client, String projectId, String instanceId, String tableId) {
+ BigtableDataClient client,
+ String projectId,
+ String instanceId,
+ String tableId,
+ Duration closeWaitTimeout) {
this.projectId = projectId;
this.instanceId = instanceId;
this.tableId = tableId;
this.bulkMutation = client.newBulkMutationBatcher(tableId);
- }
-
- @Override
- public void flush() throws IOException {
- if (bulkMutation != null) {
- try {
- stopwatch.start();
- bulkMutation.flush();
- bulkSize.update(outstandingMutations);
- outstandingMutations = 0;
- stopwatch.stop();
- latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- // We fail since flush() operation was interrupted.
- throw new IOException(e);
- }
- }
+ this.closeWaitTimeout = closeWaitTimeout;
}
@Override
public void close() throws IOException {
if (bulkMutation != null) {
try {
stopwatch.start();
- bulkMutation.flush();
- bulkMutation.close();
+ // closeAsync will send any remaining elements in the batch.
+ // If the experimental close wait timeout flag is set,
+ // set a timeout waiting for the future.
+ ApiFuture<Void> future = bulkMutation.closeAsync();
+ if (Duration.ZERO.isShorterThan(closeWaitTimeout)) {
+ future.get(closeWaitTimeout.getMillis(), TimeUnit.MILLISECONDS);
+ } else {
+ future.get();
+ }
bulkSize.update(outstandingMutations);
outstandingMutations = 0;
stopwatch.stop();
latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
+ } catch (TimeoutException e) {
+ // We fail because future.get() timed out
+ throw new IOException("BulkMutation took too long to close", e);
+ } catch (ExecutionException e) {
+ throw new IOException("ExecutionException when closing the BulkMutation", e.getCause());
Review Comment:
`new IOException("Failed to close batch", e.getCause()`
But also: doesnt the Batcher throw a BatchingException if any of its entries failed? Might want to catch that or are you expecting this behavior?
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "mutianf (via GitHub)" <gi...@apache.org>.
mutianf commented on code in PR #29548:
URL: https://github.com/apache/beam/pull/29548#discussion_r1417510336
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java:
##########
@@ -180,6 +181,27 @@
* .withBatchElements(100)); // every batch will have 100 elements
* }</pre>
*
+ * <p>Configure timeout for writes:
+ *
+ * <pre>{@code
+ * // Let each attempt run for 1 second, retry if the attempt failed.
+ * // Give up after the request is retried for 60 seconds.
+ * Duration attemptTimeout = Duration.millis(1000);
+ * Duration operationTimeout = Duration.millis(60 * 1000);
+ * data.apply("write",
+ * BigtableIO.write()
+ * .withProjectId("project")
+ * .withInstanceId("instance")
+ * .withTableId("table")
+ * .withAttemptTimeout(attemptTimeout)
Review Comment:
I think closing the service in tearDown() instead of finish_bundle() is the root cause of the hanging issue in a customer ticket, so this fix should significantly lower the chance of that happening. However, the underlying batch also seems to hang occasionally, and closing it with a timeout can mitigate that, which will cause the bundle the fail but dataflow will retry it.
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "igorbernstein2 (via GitHub)" <gi...@apache.org>.
igorbernstein2 commented on code in PR #29548:
URL: https://github.com/apache/beam/pull/29548#discussion_r1414616962
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java:
##########
@@ -1138,8 +1162,26 @@ public PCollection<BigtableWriteResult> expand(
bigtableConfig.validate();
bigtableWriteOptions.validate();
+ // Get experimental flag and set on BigtableWriteOptions
+ PipelineOptions pipelineOptions = input.getPipeline().getOptions();
+ String closeWaitTimeoutStr =
+ ExperimentalOptions.getExperimentValue(pipelineOptions, BIGTABLE_WRITER_WAIT_TIMEOUT_MS);
+ Duration closeWaitTimeout = null;
+ if (closeWaitTimeoutStr != null) {
+ long closeWaitTimeoutMs = Long.parseLong(closeWaitTimeoutStr);
+ if (closeWaitTimeoutMs < 0) {
Review Comment:
nit. Precondition.checkState
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "igorbernstein2 (via GitHub)" <gi...@apache.org>.
igorbernstein2 commented on code in PR #29548:
URL: https://github.com/apache/beam/pull/29548#discussion_r1414221675
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java:
##########
@@ -1138,8 +1162,35 @@ public PCollection<BigtableWriteResult> expand(
bigtableConfig.validate();
bigtableWriteOptions.validate();
+ // Get experimental flag and set on BigtableWriteOptions
+ PipelineOptions pipelineOptions = input.getPipeline().getOptions();
+ String closeWaitTimeoutStr =
+ ExperimentalOptions.getExperimentValue(pipelineOptions, BIGTABLE_WRITER_WAIT_TIMEOUT_MS);
+ Duration closeWaitTimeout = Duration.ZERO;
+ if (closeWaitTimeoutStr != null) {
+ try {
+ long closeWaitTimeoutMs = Long.parseLong(closeWaitTimeoutStr);
+ if (closeWaitTimeoutMs < 0) {
+ LOG.warn(
+ "Invalid close wait timeout {}, will not set a wait timeout on close",
+ closeWaitTimeoutMs);
+ } else {
+ closeWaitTimeout = Duration.millis(closeWaitTimeoutMs);
+ }
+ } catch (NumberFormatException e) {
+ LOG.warn(
Review Comment:
this should just throw
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java:
##########
@@ -1195,6 +1246,7 @@ private static class BigtableWriterFn
this.writeOptions = writeOptions;
this.failures = new ConcurrentLinkedQueue<>();
this.id = factory.newId();
+ LOG.info("Created Bigtable Write Fn with writeOptions {} ", writeOptions);
Review Comment:
I'm not sure if this should be info. WriteOptions is an internal class, so logging it by default is not useful to endusers. I would drop this to debug
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableServiceImpl.java:
##########
@@ -119,8 +121,15 @@ class BigtableServiceImpl implements BigtableService {
private final Duration readOperationTimeout;
@Override
- public BigtableWriterImpl openForWriting(String tableId) {
- return new BigtableWriterImpl(client, projectId, instanceId, tableId);
+ public BigtableWriterImpl openForWriting(BigtableWriteOptions writeOptions) {
+ return new BigtableWriterImpl(
+ client,
+ projectId,
+ instanceId,
+ writeOptions.getTableId().get(),
+ writeOptions.getCloseWaitTimeout() != null
Review Comment:
There seems to be multiple ways to express that close wait timeout should be disabled: null & 0. I think you should pick one and stick to it
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java:
##########
@@ -1138,8 +1162,35 @@ public PCollection<BigtableWriteResult> expand(
bigtableConfig.validate();
bigtableWriteOptions.validate();
+ // Get experimental flag and set on BigtableWriteOptions
+ PipelineOptions pipelineOptions = input.getPipeline().getOptions();
+ String closeWaitTimeoutStr =
+ ExperimentalOptions.getExperimentValue(pipelineOptions, BIGTABLE_WRITER_WAIT_TIMEOUT_MS);
+ Duration closeWaitTimeout = Duration.ZERO;
+ if (closeWaitTimeoutStr != null) {
+ try {
+ long closeWaitTimeoutMs = Long.parseLong(closeWaitTimeoutStr);
+ if (closeWaitTimeoutMs < 0) {
+ LOG.warn(
+ "Invalid close wait timeout {}, will not set a wait timeout on close",
+ closeWaitTimeoutMs);
Review Comment:
I think this should just throw
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableServiceImpl.java:
##########
@@ -470,50 +479,50 @@ static class BigtableWriterImpl implements Writer {
private String projectId;
private String instanceId;
private String tableId;
+ private Duration closeWaitTimeout;
private Distribution bulkSize = Metrics.distribution("BigTable-" + tableId, "batchSize");
private Distribution latency = Metrics.distribution("BigTable-" + tableId, "batchLatencyMs");
BigtableWriterImpl(
- BigtableDataClient client, String projectId, String instanceId, String tableId) {
+ BigtableDataClient client,
+ String projectId,
+ String instanceId,
+ String tableId,
+ Duration closeWaitTimeout) {
this.projectId = projectId;
this.instanceId = instanceId;
this.tableId = tableId;
this.bulkMutation = client.newBulkMutationBatcher(tableId);
- }
-
- @Override
- public void flush() throws IOException {
- if (bulkMutation != null) {
- try {
- stopwatch.start();
- bulkMutation.flush();
- bulkSize.update(outstandingMutations);
- outstandingMutations = 0;
- stopwatch.stop();
- latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- // We fail since flush() operation was interrupted.
- throw new IOException(e);
- }
- }
+ this.closeWaitTimeout = closeWaitTimeout;
}
@Override
public void close() throws IOException {
if (bulkMutation != null) {
try {
stopwatch.start();
- bulkMutation.flush();
- bulkMutation.close();
+ // closeAsync will send any remaining elements in the batch.
+ // If the experimental close wait timeout flag is set,
+ // set a timeout waiting for the future.
+ ApiFuture<Void> future = bulkMutation.closeAsync();
+ if (closeWaitTimeout.isLongerThan(Duration.ZERO)) {
+ future.get(closeWaitTimeout.getMillis(), TimeUnit.MILLISECONDS);
+ } else {
+ future.get();
+ }
bulkSize.update(outstandingMutations);
outstandingMutations = 0;
stopwatch.stop();
latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
+ } catch (TimeoutException | ExecutionException e) {
+ // We fail because future.get() timed out
+ String errorMsg = "BulkMutation took too long to close";
Review Comment:
Also I would unwrap the ExceptionxecutionExcetion cause
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableServiceImpl.java:
##########
@@ -470,50 +479,50 @@ static class BigtableWriterImpl implements Writer {
private String projectId;
private String instanceId;
private String tableId;
+ private Duration closeWaitTimeout;
private Distribution bulkSize = Metrics.distribution("BigTable-" + tableId, "batchSize");
private Distribution latency = Metrics.distribution("BigTable-" + tableId, "batchLatencyMs");
BigtableWriterImpl(
- BigtableDataClient client, String projectId, String instanceId, String tableId) {
+ BigtableDataClient client,
+ String projectId,
+ String instanceId,
+ String tableId,
+ Duration closeWaitTimeout) {
this.projectId = projectId;
this.instanceId = instanceId;
this.tableId = tableId;
this.bulkMutation = client.newBulkMutationBatcher(tableId);
- }
-
- @Override
- public void flush() throws IOException {
- if (bulkMutation != null) {
- try {
- stopwatch.start();
- bulkMutation.flush();
- bulkSize.update(outstandingMutations);
- outstandingMutations = 0;
- stopwatch.stop();
- latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- // We fail since flush() operation was interrupted.
- throw new IOException(e);
- }
- }
+ this.closeWaitTimeout = closeWaitTimeout;
}
@Override
public void close() throws IOException {
if (bulkMutation != null) {
try {
stopwatch.start();
- bulkMutation.flush();
- bulkMutation.close();
+ // closeAsync will send any remaining elements in the batch.
+ // If the experimental close wait timeout flag is set,
+ // set a timeout waiting for the future.
+ ApiFuture<Void> future = bulkMutation.closeAsync();
+ if (closeWaitTimeout.isLongerThan(Duration.ZERO)) {
+ future.get(closeWaitTimeout.getMillis(), TimeUnit.MILLISECONDS);
+ } else {
+ future.get();
+ }
bulkSize.update(outstandingMutations);
outstandingMutations = 0;
stopwatch.stop();
latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
+ } catch (TimeoutException | ExecutionException e) {
+ // We fail because future.get() timed out
+ String errorMsg = "BulkMutation took too long to close";
Review Comment:
This is not true: ExecutionException will be thrown when the future operation failed. ie if an element failed, I think it will be wrapped in an ExecutionException
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableServiceImpl.java:
##########
@@ -470,50 +479,50 @@ static class BigtableWriterImpl implements Writer {
private String projectId;
private String instanceId;
private String tableId;
+ private Duration closeWaitTimeout;
private Distribution bulkSize = Metrics.distribution("BigTable-" + tableId, "batchSize");
private Distribution latency = Metrics.distribution("BigTable-" + tableId, "batchLatencyMs");
BigtableWriterImpl(
- BigtableDataClient client, String projectId, String instanceId, String tableId) {
+ BigtableDataClient client,
+ String projectId,
+ String instanceId,
+ String tableId,
+ Duration closeWaitTimeout) {
this.projectId = projectId;
this.instanceId = instanceId;
this.tableId = tableId;
this.bulkMutation = client.newBulkMutationBatcher(tableId);
- }
-
- @Override
- public void flush() throws IOException {
- if (bulkMutation != null) {
- try {
- stopwatch.start();
- bulkMutation.flush();
- bulkSize.update(outstandingMutations);
- outstandingMutations = 0;
- stopwatch.stop();
- latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- // We fail since flush() operation was interrupted.
- throw new IOException(e);
- }
- }
+ this.closeWaitTimeout = closeWaitTimeout;
}
@Override
public void close() throws IOException {
if (bulkMutation != null) {
try {
stopwatch.start();
- bulkMutation.flush();
- bulkMutation.close();
+ // closeAsync will send any remaining elements in the batch.
+ // If the experimental close wait timeout flag is set,
+ // set a timeout waiting for the future.
+ ApiFuture<Void> future = bulkMutation.closeAsync();
+ if (closeWaitTimeout.isLongerThan(Duration.ZERO)) {
+ future.get(closeWaitTimeout.getMillis(), TimeUnit.MILLISECONDS);
+ } else {
+ future.get();
+ }
bulkSize.update(outstandingMutations);
outstandingMutations = 0;
stopwatch.stop();
latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
+ } catch (TimeoutException | ExecutionException e) {
+ // We fail because future.get() timed out
+ String errorMsg = "BulkMutation took too long to close";
+ LOG.warn(errorMsg, e);
+ throw new IOException(errorMsg, e);
Review Comment:
why log & throw? I think the exception bubbling up should be enough?
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #29548:
URL: https://github.com/apache/beam/pull/29548#discussion_r1416406133
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java:
##########
@@ -180,6 +181,27 @@
* .withBatchElements(100)); // every batch will have 100 elements
* }</pre>
*
+ * <p>Configure timeout for writes:
+ *
+ * <pre>{@code
+ * // Let each attempt run for 1 second, retry if the attempt failed.
+ * // Give up after the request is retried for 60 seconds.
+ * Duration attemptTimeout = Duration.millis(1000);
+ * Duration operationTimeout = Duration.millis(60 * 1000);
+ * data.apply("write",
+ * BigtableIO.write()
+ * .withProjectId("project")
+ * .withInstanceId("instance")
+ * .withTableId("table")
+ * .withAttemptTimeout(attemptTimeout)
Review Comment:
Just trying to understand the fix. If the timeout is not set, is the pipeline stuck could still happen? When timeout set, is it true that bundle could fail when timeout and get retried?
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "igorbernstein2 (via GitHub)" <gi...@apache.org>.
igorbernstein2 commented on code in PR #29548:
URL: https://github.com/apache/beam/pull/29548#discussion_r1416047923
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java:
##########
@@ -1237,14 +1281,7 @@ public void finishBundle(FinishBundleContext c) throws Exception {
entry.getKey().maxTimestamp(),
entry.getKey());
}
- }
- @Teardown
- public void tearDown() throws Exception {
- if (bigtableWriter != null) {
- bigtableWriter.close();
- bigtableWriter = null;
- }
if (serviceEntry != null) {
Review Comment:
can you throw this in a finally block to make sure the ref count doesnt leak?
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "mutianf (via GitHub)" <gi...@apache.org>.
mutianf commented on code in PR #29548:
URL: https://github.com/apache/beam/pull/29548#discussion_r1415791630
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableServiceImpl.java:
##########
@@ -470,50 +477,50 @@ static class BigtableWriterImpl implements Writer {
private String projectId;
private String instanceId;
private String tableId;
+ private Duration closeWaitTimeout;
private Distribution bulkSize = Metrics.distribution("BigTable-" + tableId, "batchSize");
private Distribution latency = Metrics.distribution("BigTable-" + tableId, "batchLatencyMs");
BigtableWriterImpl(
- BigtableDataClient client, String projectId, String instanceId, String tableId) {
+ BigtableDataClient client,
+ String projectId,
+ String instanceId,
+ String tableId,
+ Duration closeWaitTimeout) {
this.projectId = projectId;
this.instanceId = instanceId;
this.tableId = tableId;
this.bulkMutation = client.newBulkMutationBatcher(tableId);
- }
-
- @Override
- public void flush() throws IOException {
- if (bulkMutation != null) {
- try {
- stopwatch.start();
- bulkMutation.flush();
- bulkSize.update(outstandingMutations);
- outstandingMutations = 0;
- stopwatch.stop();
- latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- // We fail since flush() operation was interrupted.
- throw new IOException(e);
- }
- }
+ this.closeWaitTimeout = closeWaitTimeout;
}
@Override
public void close() throws IOException {
if (bulkMutation != null) {
try {
stopwatch.start();
- bulkMutation.flush();
- bulkMutation.close();
+ // closeAsync will send any remaining elements in the batch.
+ // If the experimental close wait timeout flag is set,
+ // set a timeout waiting for the future.
+ ApiFuture<Void> future = bulkMutation.closeAsync();
+ if (Duration.ZERO.isShorterThan(closeWaitTimeout)) {
+ future.get(closeWaitTimeout.getMillis(), TimeUnit.MILLISECONDS);
+ } else {
+ future.get();
+ }
bulkSize.update(outstandingMutations);
outstandingMutations = 0;
stopwatch.stop();
latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
+ } catch (TimeoutException e) {
+ // We fail because future.get() timed out
+ throw new IOException("BulkMutation took too long to close", e);
+ } catch (ExecutionException e) {
+ throw new IOException("ExecutionException when closing the BulkMutation", e.getCause());
Review Comment:
I think entry failure will be retried and if it bubbles up we rely on beam to retry the bundle?
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #29548:
URL: https://github.com/apache/beam/pull/29548#discussion_r1416406133
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java:
##########
@@ -180,6 +181,27 @@
* .withBatchElements(100)); // every batch will have 100 elements
* }</pre>
*
+ * <p>Configure timeout for writes:
+ *
+ * <pre>{@code
+ * // Let each attempt run for 1 second, retry if the attempt failed.
+ * // Give up after the request is retried for 60 seconds.
+ * Duration attemptTimeout = Duration.millis(1000);
+ * Duration operationTimeout = Duration.millis(60 * 1000);
+ * data.apply("write",
+ * BigtableIO.write()
+ * .withProjectId("project")
+ * .withInstanceId("instance")
+ * .withTableId("table")
+ * .withAttemptTimeout(attemptTimeout)
Review Comment:
Just trying to understand the fix. If the timeout is not set, is the pipeline still could stuck happen? When timeout set, is it true that bundle could fail when timeout and get retried?
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on code in PR #29548:
URL: https://github.com/apache/beam/pull/29548#discussion_r1417672665
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableIO.java:
##########
@@ -180,6 +181,27 @@
* .withBatchElements(100)); // every batch will have 100 elements
* }</pre>
*
+ * <p>Configure timeout for writes:
+ *
+ * <pre>{@code
+ * // Let each attempt run for 1 second, retry if the attempt failed.
+ * // Give up after the request is retried for 60 seconds.
+ * Duration attemptTimeout = Duration.millis(1000);
+ * Duration operationTimeout = Duration.millis(60 * 1000);
+ * data.apply("write",
+ * BigtableIO.write()
+ * .withProjectId("project")
+ * .withInstanceId("instance")
+ * .withTableId("table")
+ * .withAttemptTimeout(attemptTimeout)
Review Comment:
Thanks for the context!
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn merged PR #29548:
URL: https://github.com/apache/beam/pull/29548
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn commented on PR #29548:
URL: https://github.com/apache/beam/pull/29548#issuecomment-1841800386
Java Tests failures were a hiccup earlier today not related
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #29548:
URL: https://github.com/apache/beam/pull/29548#issuecomment-1836664379
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @Abacn for label java.
R: @chamikaramj for label io.
R: @igorbernstein2 for label bigtable.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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: [PR] fix: have a timeout on the wait when closing BigtableIO#Writer [beam]
Posted by "igorbernstein2 (via GitHub)" <gi...@apache.org>.
igorbernstein2 commented on code in PR #29548:
URL: https://github.com/apache/beam/pull/29548#discussion_r1414618247
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableServiceImpl.java:
##########
@@ -470,50 +477,50 @@ static class BigtableWriterImpl implements Writer {
private String projectId;
private String instanceId;
private String tableId;
+ private Duration closeWaitTimeout;
private Distribution bulkSize = Metrics.distribution("BigTable-" + tableId, "batchSize");
private Distribution latency = Metrics.distribution("BigTable-" + tableId, "batchLatencyMs");
BigtableWriterImpl(
- BigtableDataClient client, String projectId, String instanceId, String tableId) {
+ BigtableDataClient client,
+ String projectId,
+ String instanceId,
+ String tableId,
+ Duration closeWaitTimeout) {
this.projectId = projectId;
this.instanceId = instanceId;
this.tableId = tableId;
this.bulkMutation = client.newBulkMutationBatcher(tableId);
- }
-
- @Override
- public void flush() throws IOException {
- if (bulkMutation != null) {
- try {
- stopwatch.start();
- bulkMutation.flush();
- bulkSize.update(outstandingMutations);
- outstandingMutations = 0;
- stopwatch.stop();
- latency.update(stopwatch.elapsed(TimeUnit.MILLISECONDS));
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- // We fail since flush() operation was interrupted.
- throw new IOException(e);
- }
- }
+ this.closeWaitTimeout = closeWaitTimeout;
Review Comment:
nit, please move next to all of the other passthru assignments
--
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