Fixed data race in ProxyStream

Author: alex268

table/src/main/java/tech/ydb/table/impl/BaseSession.java

@@ -1398,6 +1398,10 @@ private abstract class ProxyStream<R, T> implements GrpcReadStream<T> {
         abstract T readValue(R message);
 
         private void onClose(Status status, Throwable th) {
+            if (result.isDone()) {
+                return;
+            }
+
             if (th != null) {
                 updateSessionState(th, null, false);
                 result.completeExceptionally(th);
@@ -1415,14 +1419,15 @@ public CompletableFuture<Status> start(Observer<T> observer) {
                 if (statusCode == StatusIds.StatusCode.SUCCESS) {
                     try {
                         observer.onNext(readValue(message));
-                    } catch (Throwable t) {
-                        result.completeExceptionally(t);
+                    } catch (Throwable th) {
+                        updateSessionState(th, null, false);
+                        result.completeExceptionally(th);
                         origin.cancel();
                     }
                 } else {
-                    Issue[] issues = Issue.fromPb(readIssues(message));
-                    StatusCode code = StatusCode.fromProto(statusCode);
-                    result.complete(Status.of(code, issues));
+                    Status status = Status.of(StatusCode.fromProto(statusCode), Issue.fromPb(readIssues(message)));
+                    updateSessionState(null, status.getCode(), false);
+                    result.complete(status);
                     origin.cancel();
                 }
             }).whenComplete(this::onClose);