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]
