io.grpc.InternalStatus#io.grpc.Status.Code源码实例Demo

@Override
public void export(
    ExportTraceServiceRequest request,
    io.grpc.stub.StreamObserver<ExportTraceServiceResponse> responseObserver) {
  receivedSpans.addAll(request.getResourceSpansList());
  responseObserver.onNext(ExportTraceServiceResponse.newBuilder().build());
  if (!returnedStatus.isOk()) {
    if (returnedStatus.getCode() == Code.DEADLINE_EXCEEDED) {
      // Do not call onCompleted to simulate a deadline exceeded.
      return;
    }
    responseObserver.onError(returnedStatus.asRuntimeException());
    return;
  }
  responseObserver.onCompleted();
}
 

@Test
public void maxMessageSizeShouldBeEnforced() throws Exception {
  // Allow the response payloads of up to 1 byte.
  startTransport(3, null, true, 1, INITIAL_WINDOW_SIZE, null);

  MockStreamListener listener = new MockStreamListener();
  OkHttpClientStream stream =
      clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
  stream.start(listener);
  stream.request(1);
  assertContainStream(3);
  frameHandler().headers(false, false, 3, 0, grpcResponseHeaders(), HeadersMode.HTTP_20_HEADERS);
  assertNotNull(listener.headers);

  // Receive the message.
  final String message = "Hello Client";
  Buffer buffer = createMessageFrame(message);
  frameHandler().data(false, 3, buffer, (int) buffer.size());

  listener.waitUntilStreamClosed();
  assertEquals(Code.RESOURCE_EXHAUSTED, listener.status.getCode());
  shutdownAndVerify();
}
 

@Override
public List<AutogenEnvironmentVariablesBlob> postVisitDeepListOfAutogenEnvironmentVariablesBlob(
    List<AutogenEnvironmentVariablesBlob> lst) throws ModelDBException {
  if (lst == null) {
    return null;
  }
  Set<String> variableNames = new HashSet<>();
  for (AutogenEnvironmentVariablesBlob blob : lst) {
    blob.postVisitDeep(this);
    variableNames.add(blob.getName());
  }
  if (variableNames.size() != lst.size()) {
    throw new ModelDBException("There are recurring variables", Code.INVALID_ARGUMENT);
  }
  return lst;
}
 

@Test
public void missingWriteQueryIsNotFound() throws IOException {
  ByteString helloWorld = ByteString.copyFromUtf8("Hello, World!");
  Digest digest = DIGEST_UTIL.compute(helloWorld);
  String uuid = UUID.randomUUID().toString();
  String resourceName = createBlobUploadResourceName(uuid, digest);

  Channel channel = InProcessChannelBuilder.forName(fakeServerName).directExecutor().build();
  ByteStreamBlockingStub service = ByteStreamGrpc.newBlockingStub(channel);

  StatusRuntimeException notFoundException = null;
  try {
    service.queryWriteStatus(
        QueryWriteStatusRequest.newBuilder().setResourceName(resourceName).build());
  } catch (StatusRuntimeException e) {
    assertThat(Status.fromThrowable(e).getCode()).isEqualTo(Code.NOT_FOUND);
    notFoundException = e;
  }
  assertThat(notFoundException).isNotNull();
}
 

@Override
public DeleteTagRequest.Response deleteTag(DeleteTagRequest request) throws ModelDBException {
  try (Session session = ModelDBHibernateUtil.getSessionFactory().openSession()) {
    RepositoryEntity repository = getRepositoryById(session, request.getRepositoryId(), true);
    TagsEntity tagsEntity =
        session.get(TagsEntity.class, new TagsEntity.TagId(request.getTag(), repository.getId()));
    if (tagsEntity == null) {
      throw new ModelDBException("Tag not found " + request.getTag(), Code.NOT_FOUND);
    }
    session.beginTransaction();
    session.delete(tagsEntity);
    session.getTransaction().commit();
    return DeleteTagRequest.Response.newBuilder().build();
  } catch (Exception ex) {
    if (ModelDBUtils.needToRetry(ex)) {
      return deleteTag(request);
    } else {
      throw ex;
    }
  }
}
 

@Test
public void resolve_addressFailure_neverLookUpServiceConfig() throws Exception {
  DnsNameResolver.enableTxt = true;
  AddressResolver mockAddressResolver = mock(AddressResolver.class);
  when(mockAddressResolver.resolveAddress(anyString()))
      .thenThrow(new IOException("no addr"));
  String name = "foo.googleapis.com";

  ResourceResolver mockResourceResolver = mock(ResourceResolver.class);
  DnsNameResolver resolver = newResolver(name, 81);
  resolver.setAddressResolver(mockAddressResolver);
  resolver.setResourceResolver(mockResourceResolver);
  resolver.start(mockListener);
  assertEquals(1, fakeExecutor.runDueTasks());
  verify(mockListener).onError(errorCaptor.capture());
  Status errorStatus = errorCaptor.getValue();
  assertThat(errorStatus.getCode()).isEqualTo(Code.UNAVAILABLE);
  assertThat(errorStatus.getCause()).hasMessageThat().contains("no addr");
  verify(mockResourceResolver, never()).resolveTxt(anyString());
}
 

@Override
public GetTagRequest.Response getTag(GetTagRequest request) throws ModelDBException {
  try (Session session = ModelDBHibernateUtil.getSessionFactory().openSession()) {
    RepositoryEntity repository = getRepositoryById(session, request.getRepositoryId());

    Query query = session.createQuery(GET_TAG_HQL);
    query.setParameter("repositoryId", repository.getId());
    query.setParameter("tag", request.getTag());
    TagsEntity tagsEntity = (TagsEntity) query.uniqueResult();
    if (tagsEntity == null) {
      throw new ModelDBException("Tag not found " + request.getTag(), Code.NOT_FOUND);
    }

    CommitEntity commitEntity = session.get(CommitEntity.class, tagsEntity.getCommit_hash());
    return GetTagRequest.Response.newBuilder().setCommit(commitEntity.toCommitProto()).build();
  } catch (Exception ex) {
    if (ModelDBUtils.needToRetry(ex)) {
      return getTag(request);
    } else {
      throw ex;
    }
  }
}
 

@Test
public void transportHeadersReceived_wrongContentType_401() {
  BaseTransportState state = new BaseTransportState(transportTracer);
  state.setListener(mockListener);
  Metadata headers = new Metadata();
  headers.put(testStatusMashaller, "401");
  headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER), "text/html");
  state.transportHeadersReceived(headers);
  state.transportDataReceived(ReadableBuffers.empty(), true);

  verify(mockListener, never()).headersRead(any(Metadata.class));
  verify(mockListener).closed(statusCaptor.capture(), same(PROCESSED), same(headers));
  assertEquals(Code.UNAUTHENTICATED, statusCaptor.getValue().getCode());
  assertTrue(statusCaptor.getValue().getDescription().contains("401"));
  assertTrue(statusCaptor.getValue().getDescription().contains("text/html"));
}
 

@Override
public void findAllInputsOutputs(
    FindAllInputsOutputs request,
    StreamObserver<FindAllInputsOutputs.Response> responseObserver) {
  QPSCountResource.inc();
  try {
    if (request.getItemsCount() == 0) {
      throw new ModelDBException("Items not specified", Code.INVALID_ARGUMENT);
    }
    try (RequestLatencyResource latencyResource =
        new RequestLatencyResource(ModelDBAuthInterceptor.METHOD_NAME.get())) {
      FindAllInputsOutputs.Response response = lineageDAO.findAllInputsOutputs(request);
      responseObserver.onNext(response);
      responseObserver.onCompleted();
    }
  } catch (Exception e) {
    ModelDBUtils.observeError(
        responseObserver, e, FindAllInputsOutputs.Response.getDefaultInstance());
  }
}
 

@Test
public void maxMessageSizeShouldBeEnforced() throws Throwable {
  startServer();
  // Allow the response payloads of up to 1 byte.
  NettyClientTransport transport = newTransport(newNegotiator(),
      1, GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE, null, true);
  callMeMaybe(transport.start(clientTransportListener));

  try {
    // Send a single RPC and wait for the response.
    new Rpc(transport).halfClose().waitForResponse();
    fail("Expected the stream to fail.");
  } catch (ExecutionException e) {
    Status status = Status.fromThrowable(e);
    assertEquals(Code.RESOURCE_EXHAUSTED, status.getCode());
    assertTrue("Missing exceeds maximum from: " + status.getDescription(),
        status.getDescription().contains("exceeds maximum"));
  }
}
 

@Test
public void transportTrailersReceived_missingStatusAfterHeadersIgnoresHttpStatus() {
  BaseTransportState state = new BaseTransportState(transportTracer);
  state.setListener(mockListener);
  Metadata headers = new Metadata();
  headers.put(testStatusMashaller, "200");
  headers.put(Metadata.Key.of("content-type", Metadata.ASCII_STRING_MARSHALLER),
      "application/grpc");
  state.transportHeadersReceived(headers);
  Metadata trailers = new Metadata();
  trailers.put(testStatusMashaller, "401");
  state.transportTrailersReceived(trailers);

  verify(mockListener).headersRead(headers);
  verify(mockListener).closed(statusCaptor.capture(), same(PROCESSED), same(trailers));
  assertEquals(Code.UNKNOWN, statusCaptor.getValue().getCode());
}
 

@Override
public void export(
    ExportMetricsServiceRequest request,
    io.grpc.stub.StreamObserver<ExportMetricsServiceResponse> responseObserver) {
  receivedMetrics.addAll(request.getResourceMetricsList());
  responseObserver.onNext(ExportMetricsServiceResponse.newBuilder().build());
  if (!returnedStatus.isOk()) {
    if (returnedStatus.getCode() == Code.DEADLINE_EXCEEDED) {
      // Do not call onCompleted to simulate a deadline exceeded.
      return;
    }
    responseObserver.onError(returnedStatus.asRuntimeException());
    return;
  }
  responseObserver.onCompleted();
}
 

@Test
void cancelAfterBegin() throws Exception {
    final StreamRecorder<StreamingInputCallResponse> responseObserver = StreamRecorder.create();
    final StreamObserver<StreamingInputCallRequest> requestObserver =
            asyncStub.streamingInputCall(responseObserver);
    requestObserver.onError(new RuntimeException());
    responseObserver.awaitCompletion();
    assertThat(responseObserver.getValues()).isEmpty();
    assertThat(GrpcStatus.fromThrowable(responseObserver.getError()).getCode()).isEqualTo(Code.CANCELLED);

    final RequestLog log = requestLogQueue.take();
    assertThat(log.isComplete()).isTrue();
    assertThat(log.responseContent()).isInstanceOf(RpcResponse.class);
    final Throwable cause = ((RpcResponse) log.responseContent()).cause();
    assertThat(cause).isInstanceOf(StatusException.class);
    assertThat(((StatusException) cause).getStatus().getCode()).isEqualTo(Code.CANCELLED);
}
 

@Test
public void maxHeaderListSizeShouldBeEnforcedOnServer() throws Exception {
  startServer(100, 1);

  NettyClientTransport transport = newTransport(newNegotiator());
  callMeMaybe(transport.start(clientTransportListener));

  try {
    // Send a single RPC and wait for the response.
    new Rpc(transport, new Metadata()).halfClose().waitForResponse();
    fail("The stream should have been failed due to server received header exceeds header list"
        + " size limit!");
  } catch (Exception e) {
    Status status = Status.fromThrowable(e);
    assertEquals(status.toString(), Status.Code.INTERNAL, status.getCode());
  }
}
 

@Test
public void maxHeaderListSizeShouldBeEnforcedOnClient() throws Exception {
  startServer();

  NettyClientTransport transport =
      newTransport(newNegotiator(), DEFAULT_MAX_MESSAGE_SIZE, 1, null, true);
  callMeMaybe(transport.start(clientTransportListener));

  try {
    // Send a single RPC and wait for the response.
    new Rpc(transport, new Metadata()).halfClose().waitForResponse();
    fail("The stream should have been failed due to client received header exceeds header list"
        + " size limit!");
  } catch (Exception e) {
    Throwable rootCause = getRootCause(e);
    Status status = ((StatusException) rootCause).getStatus();
    assertEquals(Status.Code.INTERNAL, status.getCode());
    assertEquals("HTTP/2 error code: PROTOCOL_ERROR\nReceived Rst Stream",
        status.getDescription());
  }
}
 

@Test
public void trailerNotOkWithTruncatedMessage() {
  AbstractClientStream stream =
      new BaseAbstractClientStream(allocator, statsTraceCtx, transportTracer);
  stream.start(mockListener);

  stream.request(1);
  stream.transportState().deframe(ReadableBuffers.wrap(new byte[] {0, 0, 0, 0, 2, 1}));
  stream.transportState().inboundTrailersReceived(
      new Metadata(), Status.DATA_LOSS.withDescription("data___loss"));

  ArgumentCaptor<Status> statusCaptor = ArgumentCaptor.forClass(Status.class);
  verify(mockListener)
      .closed(statusCaptor.capture(), any(RpcProgress.class), any(Metadata.class));
  assertSame(Status.Code.DATA_LOSS, statusCaptor.getValue().getCode());
  assertEquals("data___loss", statusCaptor.getValue().getDescription());
}
 

private void matchInterruptible(MatchListener listener) throws IOException, InterruptedException {
  listener.onWaitStart();
  QueueEntry queueEntry = null;
  try {
    queueEntry = backplane.dispatchOperation(platform.getPropertiesList());
  } catch (IOException e) {
    Status status = Status.fromThrowable(e);
    if (status.getCode() != Code.UNAVAILABLE) {
      throw e;
    }
    // unavailable backplane will propagate a null queueEntry
  }
  listener.onWaitEnd();
  if (queueEntry == null || satisfiesRequirements(matchProvisions, queueEntry.getPlatform())) {
    listener.onEntry(queueEntry);
  } else {
    backplane.rejectOperation(queueEntry);
  }
  if (Thread.interrupted()) {
    throw new InterruptedException();
  }
}
 

@Test
public void resolve_nullResourceResolver_addressFailure() throws Exception {
  AddressResolver mockAddressResolver = mock(AddressResolver.class);
  when(mockAddressResolver.resolveAddress(anyString())).thenThrow(new IOException("no addr"));
  ResourceResolver resourceResolver = null;

  resolver.setAddressResolver(mockAddressResolver);
  resolver.setResourceResolver(resourceResolver);

  resolver.start(mockListener);
  assertThat(fakeClock.runDueTasks()).isEqualTo(1);
  verify(mockListener).onError(errorCaptor.capture());
  Status errorStatus = errorCaptor.getValue();
  assertThat(errorStatus.getCode()).isEqualTo(Code.UNAVAILABLE);
  assertThat(errorStatus.getCause()).hasMessageThat().contains("no addr");
}
 

/**
 * Retrieves the value of a random key.
 */
private void doRetrieve(KeyValueServiceBlockingStub stub) {
  ByteString key = knownKeys.getRandomKey();
  try {
    RetrieveResponse res = stub.retrieve(RetrieveRequest.newBuilder()
        .setKey(key)
        .build());
    if (res.getValue().size() < 1) {
      throw new RuntimeException("Invalid response");
    }
  } catch (StatusRuntimeException e) {
    if (e.getStatus().getCode() == Code.NOT_FOUND) {
      knownKeys.remove(key);
      logger.log(Level.INFO, "Key not found", e);
    } else {
      throw e;
    }
  }
}
 

/**
 * Updates a random key with a random value.
 */
private void doUpdate(KeyValueServiceBlockingStub stub) {
  ByteString key = knownKeys.getRandomKey();
  try {
    UpdateResponse res = stub.update(UpdateRequest.newBuilder()
        .setKey(key)
        .setValue(randomBytes(MEAN_VALUE_SIZE))
        .build());
    if (!res.equals(UpdateResponse.getDefaultInstance())) {
      throw new RuntimeException("Invalid response");
    }
  } catch (StatusRuntimeException e) {
    if (e.getStatus().getCode() == Code.NOT_FOUND) {
      knownKeys.remove(key);
      logger.log(Level.INFO, "Key not found", e);
    } else {
      throw e;
    }
  }
}
 

static Blob getBlob(Blob.ContentCase contentCase)
    throws ModelDBException, NoSuchAlgorithmException {
  switch (contentCase) {
    case DATASET:
      DatasetBlob datasetBlob = DatasetBlob.newBuilder().setPath(getPathDatasetBlob()).build();
      return Blob.newBuilder().setDataset(datasetBlob).build();
    case CODE:
      return getCodeBlobFromPath("abc");
    case ENVIRONMENT:
      break;
    case CONFIG:
      List<HyperparameterConfigBlob> hyperparameterConfigBlobs = getHyperparameterConfigList();
      List<HyperparameterSetConfigBlob> setConfigBlobs = getContinuesList();
      ConfigBlob configBlob =
          ConfigBlob.newBuilder()
              .addAllHyperparameters(hyperparameterConfigBlobs)
              .addAllHyperparameterSet(setConfigBlobs)
              .build();
      return Blob.newBuilder().setConfig(configBlob).build();
    case CONTENT_NOT_SET:
    default:
      throw new ModelDBException("Invalid blob type found", Status.Code.INVALID_ARGUMENT);
  }
  throw new ModelDBException("Invalid blob type found", Status.Code.INVALID_ARGUMENT);
}
 

@Override
public Blob getBlob(Session session) throws ModelDBException {
  Builder codeBlobBuilder = CodeBlob.newBuilder();
  switch (getElementType()) {
    case GIT_CODE_BLOB:
      codeBlobBuilder.setGit(session.get(GitCodeBlobEntity.class, getElementSha()).toProto());
      break;
    case NOTEBOOK_CODE_BLOB:
      NotebookCodeBlobEntity notebookCodeBlobEntity =
          session.get(NotebookCodeBlobEntity.class, getElementSha());
      String datasetBlobHash = notebookCodeBlobEntity.getPath_dataset_blob_hash();
      final NotebookCodeBlob.Builder builder = NotebookCodeBlob.newBuilder();
      PathDatasetBlob pathBlob = DatasetBlobFactory.getPathBlob(session, datasetBlobHash);
      if (pathBlob != null) {
        if (pathBlob.getComponentsCount() == 1) {
          builder.setPath(pathBlob.getComponents(0));
        } else {
          throw new ModelDBException("Path should have only one component", Code.INTERNAL);
        }
      }
      codeBlobBuilder.setNotebook(
          builder.setGitRepo(notebookCodeBlobEntity.getGitCodeBlobEntity().toProto()).build());
      break;
  }
  return Blob.newBuilder().setCode(codeBlobBuilder).build();
}
 

@Override
public boolean tableExists(String tableId) throws IOException {
  try (BigtableSession session = new BigtableSession(options)) {
    GetTableRequest getTable =
        GetTableRequest.newBuilder()
            .setName(options.getInstanceName().toTableNameStr(tableId))
            .build();
    session.getTableAdminClient().getTable(getTable);
    return true;
  } catch (StatusRuntimeException e) {
    if (e.getStatus().getCode() == Code.NOT_FOUND) {
      return false;
    }
    String message =
        String.format(
            "Error checking whether table %s (BigtableOptions %s) exists", tableId, options);
    LOG.error(message, e);
    throw new IOException(message, e);
  }
}
 

/**
 * Creates one or more media items in a user's Google Photos library and adds them to an album.
 *
 * @param albumId Identifier of the album where the media items are added. The media items are
 *     also added to the user's library.
 * @param newMediaItems List of media items to be created.
 * @see #batchCreateMediaItems(String, List, AlbumPosition)
 */
public final BatchCreateMediaItemsResponse batchCreateMediaItems(
    String albumId, List<NewMediaItem> newMediaItems) {
  if (Strings.isNullOrEmpty(albumId)) {
    throw new InvalidArgumentException(
        "Request must have an album id.",
        null /* cause */,
        GrpcStatusCode.of(Code.INVALID_ARGUMENT),
        false /* retryable */);
  }

  BatchCreateMediaItemsRequest request =
      BatchCreateMediaItemsRequest.newBuilder()
          .setAlbumId(albumId)
          .addAllNewMediaItems(newMediaItems)
          .build();
  return batchCreateMediaItems(request);
}
 

@Override
public Blob getBlob(Session session) throws ModelDBException {
  DatasetBlob.Builder datasetBlobBuilder = DatasetBlob.newBuilder();
  switch (getElementType()) {
    case S_3_DATASET_BLOB:
      return Blob.newBuilder()
          .setDataset(datasetBlobBuilder.setS3(getS3Blob(session, getElementSha())))
          .build();
    case PATH_DATASET_BLOB:
      final PathDatasetBlob pathBlob = getPathBlob(session, getElementSha());
      if (pathBlob == null) {
        throw new ModelDBException("Path blob not found", Code.INTERNAL);
      }
      return Blob.newBuilder().setDataset(datasetBlobBuilder.setPath(pathBlob)).build();
  }
  return Blob.newBuilder().setDataset(datasetBlobBuilder).build();
}
 

@Override
public void rstStream(int streamId, ErrorCode errorCode) {
  logger.logRstStream(OkHttpFrameLogger.Direction.INBOUND, streamId, errorCode);
  Status status = toGrpcStatus(errorCode).augmentDescription("Rst Stream");
  boolean stopDelivery =
      (status.getCode() == Code.CANCELLED || status.getCode() == Code.DEADLINE_EXCEEDED);
  synchronized (lock) {
    OkHttpClientStream stream = streams.get(streamId);
    if (stream != null) {
      PerfMark.event("OkHttpClientTransport$ClientFrameHandler.rstStream",
          stream.transportState().tag());
      finishStream(
          streamId, status,
          errorCode == ErrorCode.REFUSED_STREAM ? RpcProgress.REFUSED : RpcProgress.PROCESSED,
          stopDelivery, null, null);
    }
  }
}
 

@Test
public void maxMessageSizeShouldBeEnforced() throws Exception {
  // Allow the response payloads of up to 1 byte.
  startTransport(3, null, true, 1, INITIAL_WINDOW_SIZE, null);

  MockStreamListener listener = new MockStreamListener();
  OkHttpClientStream stream =
      clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
  stream.start(listener);
  stream.request(1);
  assertContainStream(3);
  frameHandler().headers(false, false, 3, 0, grpcResponseHeaders(), HeadersMode.HTTP_20_HEADERS);
  assertNotNull(listener.headers);

  // Receive the message.
  final String message = "Hello Client";
  Buffer buffer = createMessageFrame(message);
  frameHandler().data(false, 3, buffer, (int) buffer.size());

  listener.waitUntilStreamClosed();
  assertEquals(Code.RESOURCE_EXHAUSTED, listener.status.getCode());
  shutdownAndVerify();
}
 

@Override
public void findAllInputs(
    FindAllInputs request, StreamObserver<FindAllInputs.Response> responseObserver) {
  QPSCountResource.inc();
  try {
    if (request.getItemsCount() == 0) {
      throw new ModelDBException("Items not specified", Code.INVALID_ARGUMENT);
    }
    try (RequestLatencyResource latencyResource =
        new RequestLatencyResource(ModelDBAuthInterceptor.METHOD_NAME.get())) {
      FindAllInputs.Response response = lineageDAO.findAllInputs(request);
      responseObserver.onNext(response);
      responseObserver.onCompleted();
    }
  } catch (Exception e) {
    ModelDBUtils.observeError(responseObserver, e, FindAllInputs.Response.getDefaultInstance());
  }
}
 

@Override
public void addLineage(AddLineage request, StreamObserver<AddLineage.Response> responseObserver) {
  QPSCountResource.inc();
  try {
    if (request.getInputCount() == 0 && request.getOutputCount() == 0) {
      throw new ModelDBException("Input and output not specified", Code.INVALID_ARGUMENT);
    } else {
      if (request.getInputCount() == 0) {
        throw new ModelDBException("Input not specified", Code.INVALID_ARGUMENT);
      } else if (request.getOutputCount() == 0) {
        throw new ModelDBException("Output not specified", Code.INVALID_ARGUMENT);
      }
    }
    try (RequestLatencyResource latencyResource =
        new RequestLatencyResource(ModelDBAuthInterceptor.METHOD_NAME.get())) {
      AddLineage.Response response = lineageDAO.addLineage(request, this::isResourceExists);
      responseObserver.onNext(response);
      responseObserver.onCompleted();
    }
  } catch (Exception e) {
    ModelDBUtils.observeError(responseObserver, e, AddLineage.Response.getDefaultInstance());
  }
}
 

@Test
public void testSecretWatcher_onSecretChangedException_expectNack() throws IOException {
  SdsClient.SecretWatcher mockWatcher = mock(SdsClient.SecretWatcher.class);

  when(serverMock.getSecretFor("name1"))
      .thenReturn(getOneTlsCertSecret("name1", SERVER_0_KEY_FILE, SERVER_0_PEM_FILE));
  doThrow(new RuntimeException("test exception-abc"))
      .when(mockWatcher)
      .onSecretChanged(any(Secret.class));

  sdsClient.watchSecret(mockWatcher);
  verifyDiscoveryRequest(server.lastGoodRequest, "", "", node, "name1");
  assertThat(server.lastRequestOnlyForAck).isNull();
  assertThat(server.lastNack).isNotNull();
  assertThat(server.lastNack.getVersionInfo()).isEmpty();
  assertThat(server.lastNack.getResponseNonce()).isEmpty();
  com.google.rpc.Status errorDetail = server.lastNack.getErrorDetail();
  assertThat(errorDetail.getCode()).isEqualTo(Code.UNKNOWN.value());
  assertThat(errorDetail.getMessage()).isEqualTo("Secret not updated");
}