@Test public void testBatchDmlErrorPropagation() throws IOException { ResultSet expectedResultSet = ResultSet.newBuilder() .setStats(ResultSetStats.newBuilder().setRowCountExact(20)) .build(); doTest(new SpannerImplBase() { @Override public void executeBatchDml( ExecuteBatchDmlRequest request, StreamObserver<ExecuteBatchDmlResponse> responseObserver) { ExecuteBatchDmlResponse response = ExecuteBatchDmlResponse.newBuilder() .setStatus( Status.newBuilder() .setCode(Code.ABORTED.getHttpStatusCode()) .setMessage("error message")) .addResultSets(expectedResultSet) .build(); responseObserver.onNext(response); responseObserver.onCompleted(); } }, grpcClient -> StepVerifier .create(grpcClient.executeBatchDml(this.mockContext, new ArrayList<>())) .expectNext(expectedResultSet) .expectErrorMessage("error message") .verify() ); }
@Override public Flux<ResultSet> executeBatchDml( StatementExecutionContext ctx, List<Statement> statements) { return Mono.defer(() -> { ExecuteBatchDmlRequest.Builder request = ExecuteBatchDmlRequest.newBuilder() .setSession(ctx.getSessionName()) .addAllStatements(statements) .setSeqno(ctx.nextSeqNum()); if (ctx.getTransactionId() != null) { request.setTransaction(TransactionSelector.newBuilder().setId(ctx.getTransactionId())); return ObservableReactiveUtil.unaryCall( obs -> this.spanner.executeBatchDml(request.build(), obs)); } else { request.setTransaction(TransactionSelector.newBuilder().setBegin(READ_WRITE_TRANSACTION)); return ObservableReactiveUtil .<ExecuteBatchDmlResponse>unaryCall( obs -> this.spanner.executeBatchDml(request.build(), obs)) .delayUntil(response -> { if (response.getResultSetsList().size() > 0) { Transaction transaction = response.getResultSets(0) .getMetadata() .getTransaction(); return commitTransaction(ctx.getSessionName(), transaction); } else { return Mono.empty(); } }); } }) .flatMapMany(response -> { Flux<ResultSet> results = Flux.fromIterable(response.getResultSetsList()); if (response.hasStatus() && response.getStatus().getCode() != Status.Code.OK.value()) { R2dbcException exception = SpannerExceptionUtil.createR2dbcException( response.getStatus().getCode(), response.getStatus().getMessage()); results = results.concatWith(Mono.error(exception)); } return results; }); }
GrpcClient implements Client { @Override public Flux<ResultSet> executeBatchDml( StatementExecutionContext ctx, List<Statement> statements) { return Mono.defer(() -> { ExecuteBatchDmlRequest.Builder request = ExecuteBatchDmlRequest.newBuilder() .setSession(ctx.getSessionName()) .addAllStatements(statements) .setSeqno(ctx.nextSeqNum()); if (ctx.getTransactionId() != null) { request.setTransaction(TransactionSelector.newBuilder().setId(ctx.getTransactionId())); return ObservableReactiveUtil.unaryCall( obs -> this.spanner.executeBatchDml(request.build(), obs)); } else { request.setTransaction(TransactionSelector.newBuilder().setBegin(READ_WRITE_TRANSACTION)); return ObservableReactiveUtil .<ExecuteBatchDmlResponse>unaryCall( obs -> this.spanner.executeBatchDml(request.build(), obs)) .delayUntil(response -> { if (response.getResultSetsList().size() > 0) { Transaction transaction = response.getResultSets(0) .getMetadata() .getTransaction(); return commitTransaction(ctx.getSessionName(), transaction); } else { return Mono.empty(); } }); } }) .flatMapMany(response -> { Flux<ResultSet> results = Flux.fromIterable(response.getResultSetsList()); if (response.hasStatus() && response.getStatus().getCode() != Status.Code.OK.value()) { R2dbcException exception = SpannerExceptionUtil.createR2dbcException( response.getStatus().getCode(), response.getStatus().getMessage()); results = results.concatWith(Mono.error(exception)); } return results; }); } }
GrpcClient implements Client { @Override public Flux<ResultSet> executeBatchDml( StatementExecutionContext ctx, List<Statement> statements) { return Mono.defer(() -> { ExecuteBatchDmlRequest.Builder request = ExecuteBatchDmlRequest.newBuilder() .setSession(ctx.getSessionName()) .addAllStatements(statements) .setSeqno(ctx.nextSeqNum()); if (ctx.getTransactionId() != null) { request.setTransaction(TransactionSelector.newBuilder().setId(ctx.getTransactionId())); return ObservableReactiveUtil.unaryCall( obs -> this.spanner.executeBatchDml(request.build(), obs)); } else { request.setTransaction(TransactionSelector.newBuilder().setBegin(READ_WRITE_TRANSACTION)); return ObservableReactiveUtil .<ExecuteBatchDmlResponse>unaryCall( obs -> this.spanner.executeBatchDml(request.build(), obs)) .delayUntil(response -> { if (response.getResultSetsList().size() > 0) { Transaction transaction = response.getResultSets(0) .getMetadata() .getTransaction(); return commitTransaction(ctx.getSessionName(), transaction); } else { return Mono.empty(); } }); } }) .flatMapMany(response -> { Flux<ResultSet> results = Flux.fromIterable(response.getResultSetsList()); if (response.hasStatus() && response.getStatus().getCode() != Status.Code.OK.value()) { R2dbcException exception = SpannerExceptionUtil.createR2dbcException( response.getStatus().getCode(), response.getStatus().getMessage()); results = results.concatWith(Mono.error(exception)); } return results; }); } GrpcClient(GoogleCredentials credentials); @VisibleForTesting  GrpcClient(SpannerStub spanner, DatabaseAdminStub databaseAdmin, OperationsStub operations); }
GrpcClient implements Client { @Override public Flux<ResultSet> executeBatchDml( StatementExecutionContext ctx, List<Statement> statements) { return Mono.defer(() -> { ExecuteBatchDmlRequest.Builder request = ExecuteBatchDmlRequest.newBuilder() .setSession(ctx.getSessionName()) .addAllStatements(statements) .setSeqno(ctx.nextSeqNum()); if (ctx.getTransactionId() != null) { request.setTransaction(TransactionSelector.newBuilder().setId(ctx.getTransactionId())); return ObservableReactiveUtil.unaryCall( obs -> this.spanner.executeBatchDml(request.build(), obs)); } else { request.setTransaction(TransactionSelector.newBuilder().setBegin(READ_WRITE_TRANSACTION)); return ObservableReactiveUtil .<ExecuteBatchDmlResponse>unaryCall( obs -> this.spanner.executeBatchDml(request.build(), obs)) .delayUntil(response -> { if (response.getResultSetsList().size() > 0) { Transaction transaction = response.getResultSets(0) .getMetadata() .getTransaction(); return commitTransaction(ctx.getSessionName(), transaction); } else { return Mono.empty(); } }); } }) .flatMapMany(response -> { Flux<ResultSet> results = Flux.fromIterable(response.getResultSetsList()); if (response.hasStatus() && response.getStatus().getCode() != Status.Code.OK.value()) { R2dbcException exception = SpannerExceptionUtil.createR2dbcException( response.getStatus().getCode(), response.getStatus().getMessage()); results = results.concatWith(Mono.error(exception)); } return results; }); } GrpcClient(GoogleCredentials credentials); @VisibleForTesting  GrpcClient(SpannerStub spanner, DatabaseAdminStub databaseAdmin, OperationsStub operations); @Override Mono<Transaction> beginTransaction(
      String sessionName, TransactionOptions transactionOptions); @Override Mono<CommitResponse> commitTransaction(String sessionName, Transaction transaction); @Override Mono<Void> rollbackTransaction(String sessionName, Transaction transaction); @Override Mono<Session> createSession(String databaseName); @Override Mono<Void> deleteSession(String sessionName); @Override Flux<ResultSet> executeBatchDml(
      StatementExecutionContext ctx,
      List<Statement> statements); @Override Flux<PartialResultSet> executeStreamingSql(
      StatementExecutionContext ctx,
      String sql,
      Struct params,
      Map<String, Type> types); @Override Mono<Operation> executeDdl(
      String fullyQualifiedDatabaseName,
      List<String> ddlStatements,
      Duration ddlOperationTimeout,
      Duration ddlPollInterval); @Override Mono<Void> close(); @Override Mono<Boolean> healthcheck(StatementExecutionContext ctx); @VisibleForTesting SpannerStub getSpanner(); }
GrpcClient implements Client { @Override public Flux<ResultSet> executeBatchDml( StatementExecutionContext ctx, List<Statement> statements) { return Mono.defer(() -> { ExecuteBatchDmlRequest.Builder request = ExecuteBatchDmlRequest.newBuilder() .setSession(ctx.getSessionName()) .addAllStatements(statements) .setSeqno(ctx.nextSeqNum()); if (ctx.getTransactionId() != null) { request.setTransaction(TransactionSelector.newBuilder().setId(ctx.getTransactionId())); return ObservableReactiveUtil.unaryCall( obs -> this.spanner.executeBatchDml(request.build(), obs)); } else { request.setTransaction(TransactionSelector.newBuilder().setBegin(READ_WRITE_TRANSACTION)); return ObservableReactiveUtil .<ExecuteBatchDmlResponse>unaryCall( obs -> this.spanner.executeBatchDml(request.build(), obs)) .delayUntil(response -> { if (response.getResultSetsList().size() > 0) { Transaction transaction = response.getResultSets(0) .getMetadata() .getTransaction(); return commitTransaction(ctx.getSessionName(), transaction); } else { return Mono.empty(); } }); } }) .flatMapMany(response -> { Flux<ResultSet> results = Flux.fromIterable(response.getResultSetsList()); if (response.hasStatus() && response.getStatus().getCode() != Status.Code.OK.value()) { R2dbcException exception = SpannerExceptionUtil.createR2dbcException( response.getStatus().getCode(), response.getStatus().getMessage()); results = results.concatWith(Mono.error(exception)); } return results; }); } GrpcClient(GoogleCredentials credentials); @VisibleForTesting  GrpcClient(SpannerStub spanner, DatabaseAdminStub databaseAdmin, OperationsStub operations); @Override Mono<Transaction> beginTransaction(
      String sessionName, TransactionOptions transactionOptions); @Override Mono<CommitResponse> commitTransaction(String sessionName, Transaction transaction); @Override Mono<Void> rollbackTransaction(String sessionName, Transaction transaction); @Override Mono<Session> createSession(String databaseName); @Override Mono<Void> deleteSession(String sessionName); @Override Flux<ResultSet> executeBatchDml(
      StatementExecutionContext ctx,
      List<Statement> statements); @Override Flux<PartialResultSet> executeStreamingSql(
      StatementExecutionContext ctx,
      String sql,
      Struct params,
      Map<String, Type> types); @Override Mono<Operation> executeDdl(
      String fullyQualifiedDatabaseName,
      List<String> ddlStatements,
      Duration ddlOperationTimeout,
      Duration ddlPollInterval); @Override Mono<Void> close(); @Override Mono<Boolean> healthcheck(StatementExecutionContext ctx); @VisibleForTesting SpannerStub getSpanner();  }