@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 shouldBuildAuthHeader() throws Exception {
    // Given:
    givenCustomBasicAuthHeader();
    givenValidCredentialsFile();

    // When:
    final DefaultConnectClient connectClient =
        connectClientFactory.get(Optional.empty(), Collections.emptyList(), Optional.empty());

    // Then:
    assertThat(connectClient.getRequestHeaders(),
        arrayContaining(header(AUTH_HEADER_NAME, EXPECTED_HEADER)));
  }