@VisibleForTesting
  static SpannerOptions buildSpannerOptions(SpannerConfig spannerConfig) {
    SpannerOptions.Builder builder = SpannerOptions.newBuilder();

    Set<Code> retryableCodes = new HashSet<>();
    if (spannerConfig.getRetryableCodes() != null) {
      retryableCodes.addAll(spannerConfig.getRetryableCodes());
    }
    if (spannerConfig.getDataBoostEnabled() != null && spannerConfig.getDataBoostEnabled().get()) {
      retryableCodes.add(Code.RESOURCE_EXHAUSTED);
    }
    // Add default retryable codes for unary methods
    Set<Code> unaryMethodRetryableCodes = new HashSet<>(retryableCodes);
    unaryMethodRetryableCodes.addAll(
        builder.getSpannerStubSettingsBuilder().getSessionSettings().getRetryableCodes());
    // Set retryable codes for all API methods
    builder
        .getSpannerStubSettingsBuilder()
        .applyToAllUnaryMethods(
            input -> {
              input.setRetryableCodes(unaryMethodRetryableCodes);
              return null;
            });
    // Add default retryable codes for streaming methods
    Set<Code> streamingMethodRetryableCodes = new HashSet<>(retryableCodes);
    streamingMethodRetryableCodes.addAll(
        builder.getSpannerStubSettingsBuilder().executeStreamingSqlSettings().getRetryableCodes());
    builder
        .getSpannerStubSettingsBuilder()
        .executeStreamingSqlSettings()
        .setRetryableCodes(streamingMethodRetryableCodes);
    builder
        .getSpannerStubSettingsBuilder()
        .streamingReadSettings()
        .setRetryableCodes(streamingMethodRetryableCodes);

    // Set commit retry settings
    UnaryCallSettings.Builder<CommitRequest, CommitResponse> commitSettings =
        builder.getSpannerStubSettingsBuilder().commitSettings();
    ValueProvider<Duration> commitDeadline = spannerConfig.getCommitDeadline();
    if (spannerConfig.getCommitRetrySettings() != null) {
      commitSettings.setRetrySettings(spannerConfig.getCommitRetrySettings());
    } else if (commitDeadline != null && commitDeadline.get().getMillis() > 0) {
      // Set the GRPC deadline on the Commit API call.
      RetrySettings.Builder commitRetrySettingsBuilder =
          commitSettings.getRetrySettings().toBuilder();
      commitSettings.setRetrySettings(
          commitRetrySettingsBuilder
              .setTotalTimeout(org.threeten.bp.Duration.ofMillis(commitDeadline.get().getMillis()))
              .setMaxRpcTimeout(org.threeten.bp.Duration.ofMillis(commitDeadline.get().getMillis()))
              .setInitialRpcTimeout(
                  org.threeten.bp.Duration.ofMillis(commitDeadline.get().getMillis()))
              .build());
    }

    // Set execute streaming sql retry settings
    ServerStreamingCallSettings.Builder<ExecuteSqlRequest, PartialResultSet>
        executeStreamingSqlSettings =
            builder.getSpannerStubSettingsBuilder().executeStreamingSqlSettings();
    if (spannerConfig.getExecuteStreamingSqlRetrySettings() != null) {
      executeStreamingSqlSettings.setRetrySettings(
          spannerConfig.getExecuteStreamingSqlRetrySettings());
    } else {
      // Setting the timeout for streaming read to 2 hours. This is 1 hour by default
      // after BEAM 2.20.
      RetrySettings.Builder executeSqlStreamingRetrySettings =
          executeStreamingSqlSettings.getRetrySettings().toBuilder();
      executeStreamingSqlSettings.setRetrySettings(
          executeSqlStreamingRetrySettings
              .setInitialRpcTimeout(org.threeten.bp.Duration.ofMinutes(120))
              .setMaxRpcTimeout(org.threeten.bp.Duration.ofMinutes(120))
              .setTotalTimeout(org.threeten.bp.Duration.ofMinutes(120))
              .build());
    }

    SpannerStubSettings.Builder spannerStubSettingsBuilder =
        builder.getSpannerStubSettingsBuilder();
    ValueProvider<Duration> partitionQueryTimeout = spannerConfig.getPartitionQueryTimeout();
    if (partitionQueryTimeout != null
        && partitionQueryTimeout.get() != null
        && partitionQueryTimeout.get().getMillis() > 0) {
      spannerStubSettingsBuilder
          .partitionQuerySettings()
          .setSimpleTimeoutNoRetries(
              org.threeten.bp.Duration.ofMillis(partitionQueryTimeout.get().getMillis()));
    }
    ValueProvider<Duration> partitionReadTimeout = spannerConfig.getPartitionReadTimeout();
    if (partitionReadTimeout != null
        && partitionReadTimeout.get() != null
        && partitionReadTimeout.get().getMillis() > 0) {
      spannerStubSettingsBuilder
          .partitionReadSettings()
          .setSimpleTimeoutNoRetries(
              org.threeten.bp.Duration.ofMillis(partitionReadTimeout.get().getMillis()));
    }

    ValueProvider<String> projectId = spannerConfig.getProjectId();
    if (projectId != null) {
      builder.setProjectId(projectId.get());
    }
    ServiceFactory<Spanner, SpannerOptions> serviceFactory = spannerConfig.getServiceFactory();
    if (serviceFactory != null) {
      builder.setServiceFactory(serviceFactory);
    }
    ValueProvider<String> host = spannerConfig.getHost();
    if (host != null) {
      builder.setHost(host.get());
    }
    ValueProvider<String> emulatorHost = spannerConfig.getEmulatorHost();
    if (emulatorHost != null) {
      builder.setEmulatorHost(emulatorHost.get());
      if (spannerConfig.getIsLocalChannelProvider() != null
          && spannerConfig.getIsLocalChannelProvider().get()) {
        builder.setChannelProvider(LocalChannelProvider.create(emulatorHost.get()));
      }
      builder.setCredentials(NoCredentials.getInstance());
    }
    String userAgentString = USER_AGENT_PREFIX + "/" + ReleaseInfo.getReleaseInfo().getVersion();
    builder.setHeaderProvider(FixedHeaderProvider.create("user-agent", userAgentString));
    ValueProvider<String> databaseRole = spannerConfig.getDatabaseRole();
    if (databaseRole != null && databaseRole.get() != null && !databaseRole.get().isEmpty()) {
      builder.setDatabaseRole(databaseRole.get());
    }
    ValueProvider<Credentials> credentials = spannerConfig.getCredentials();
    if (credentials != null && credentials.get() != null) {
      builder.setCredentials(credentials.get());
    }
    SessionPoolOptions sessionPoolOptions =
        SessionPoolOptions.newBuilder().setFailIfPoolExhausted().build();
    builder.setSessionPoolOption(sessionPoolOptions);
    return builder.build();
  }
@Test
  public void testBuildSpannerOptionsWithCredential() {
    TestCredential testCredential = new TestCredential();
    SpannerConfig config1 =
        SpannerConfig.create()
            .toBuilder()
            .setServiceFactory(serviceFactory)
            .setProjectId(StaticValueProvider.of("project"))
            .setInstanceId(StaticValueProvider.of("test-instance"))
            .setDatabaseId(StaticValueProvider.of("test-db"))
            .setDatabaseRole(StaticValueProvider.of("test-role"))
            .setCredentials(StaticValueProvider.of(testCredential))
            .build();

    SpannerOptions options = SpannerAccessor.buildSpannerOptions(config1);
    assertEquals("project", options.getProjectId());
    assertEquals("test-role", options.getDatabaseRole());
    assertEquals(testCredential, options.getCredentials());
    assertNotNull(options.getSessionPoolOptions());
  }