@Override
  public synchronized DefaultConnectClient get(
      final Optional<String> ksqlAuthHeader,
      final List<Entry<String, String>> incomingRequestHeaders,
      final Optional<KsqlPrincipal> userPrincipal
  ) {
    if (defaultConnectAuthHeader == null) {
      defaultConnectAuthHeader = buildDefaultAuthHeader();
    }

    final Map<String, Object> configWithPrefixOverrides =
        ksqlConfig.valuesWithPrefixOverride(KsqlConfig.KSQL_CONNECT_PREFIX);

    return new DefaultConnectClient(
        ksqlConfig.getString(KsqlConfig.CONNECT_URL_PROPERTY),
        buildAuthHeader(ksqlAuthHeader, incomingRequestHeaders),
        requestHeadersExtension
            .map(extension -> extension.getHeaders(userPrincipal))
            .orElse(Collections.emptyMap()),
        Optional.ofNullable(newSslContext(configWithPrefixOverrides)),
        shouldVerifySslHostname(configWithPrefixOverrides),
        ksqlConfig.getLong(KsqlConfig.CONNECT_REQUEST_TIMEOUT_MS)
    );
  }
@Test
  public void shouldBuildAuthHeaderOnlyOnce() throws Exception {
    // Given:
    givenCustomBasicAuthHeader();
    givenValidCredentialsFile();

    // When: get() is called twice
    connectClientFactory.get(Optional.empty(), Collections.emptyList(), Optional.empty());
    connectClientFactory.get(Optional.empty(), Collections.emptyList(), Optional.empty());

    // Then: only loaded the credentials once -- ideally we'd check the number of times the file
    //       was read but this is an acceptable proxy for this unit test
    verify(config, times(1)).getString(KsqlConfig.CONNECT_BASIC_AUTH_CREDENTIALS_FILE_PROPERTY);
  }