[
https://issues.apache.org/jira/browse/BEAM-13402?focusedWorklogId=701703&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-701703
]
ASF GitHub Bot logged work on BEAM-13402:
-----------------------------------------
Author: ASF GitHub Bot
Created on: 28/Dec/21 21:16
Start Date: 28/Dec/21 21:16
Worklog Time Spent: 10m
Work Description: dpcollins-google commented on a change in pull request
#16215:
URL: https://github.com/apache/beam/pull/16215#discussion_r776075118
##########
File path:
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsublite/internal/PubsubLiteSink.java
##########
@@ -46,98 +41,44 @@
private final PublisherOptions options;
@GuardedBy("this")
- private transient PublisherOrError publisherOrError;
-
- // Whenever outstanding is decremented, notify() must be called.
- @GuardedBy("this")
- private transient int outstanding;
-
- @GuardedBy("this")
- private transient Deque<CheckedApiException> errorsSinceLastFinish;
+ private transient RunState runState;
public PubsubLiteSink(PublisherOptions options) {
this.options = options;
}
- @Setup
- public void setup() throws ApiException {
- Publisher<MessageMetadata> publisher;
- publisher = PerServerPublisherCache.PUBLISHER_CACHE.get(options);
- synchronized (this) {
- outstanding = 0;
- errorsSinceLastFinish = new ArrayDeque<>();
- publisherOrError = PublisherOrError.ofPublisher(publisher);
+ private static class RunState {
+ private final Deque<ApiFuture<MessageMetadata>> futures = new
ArrayDeque<>();
+
+ private final Publisher<MessageMetadata> publisher;
+
+ RunState(PublisherOptions options) {
+ publisher = PerServerPublisherCache.PUBLISHER_CACHE.get(options);
+ }
+
+ void publish(PubSubMessage message) {
+ futures.add(publisher.publish(Message.fromProto(message)));
+ }
+
+ void waitForDone() throws Exception {
Review comment:
It is important that if any exception is thrown, the bundle closure
fails, since the publishes need to be retried. I could add logging, but the
exception still needs to be rethrown.
--
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]
Issue Time Tracking
-------------------
Worklog Id: (was: 701703)
Time Spent: 11h (was: 10h 50m)
> Pub/Sub Lite commits can take 10s of hours without completing
> -------------------------------------------------------------
>
> Key: BEAM-13402
> URL: https://issues.apache.org/jira/browse/BEAM-13402
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp
> Reporter: Daniel Collins
> Priority: P2
> Time Spent: 11h
> Remaining Estimate: 0h
>
> [https://github.com/googleapis/gax-java/issues/1577] filed upstream
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
