@Override
  public ConnectResponse<ConfigInfos> validate(
      final String plugin,
      final Map<String, String> config) {
    try {
      final Map<String, String> maskedConfig = QueryMask.getMaskedConnectConfig(config);
      LOG.debug("Issuing validate request to Kafka Connect at URI {} for plugin {} and config {}",
          connectUri,
          plugin,
          maskedConfig);

      final ConnectResponse<ConfigInfos> connectResponse = withRetries(() -> Request
          .put(resolveUri(String.format(VALIDATE_CONNECTOR, plugin)))
          .setHeaders(requestHeaders)
          .responseTimeout(Timeout.ofMilliseconds(requestTimeoutMs))
          .connectTimeout(Timeout.ofMilliseconds(requestTimeoutMs))
          .bodyString(MAPPER.writeValueAsString(config), ContentType.APPLICATION_JSON)
          .execute(httpClient)
          .handleResponse(
              createHandler(HttpStatus.SC_OK, new TypeReference<ConfigInfos>() {},
                  Function.identity())));

      connectResponse.error()
          .ifPresent(error ->
              LOG.warn("Did not VALIDATE connector configuration for plugin {} and config {}: {}",
              plugin, maskedConfig, error));

      return connectResponse;

    } catch (final Exception e) {
      throw new KsqlServerException(e);
    }
  }
@Test
  public void testValidate() throws JsonProcessingException {
    // Given:
    final String plugin = SAMPLE_PLUGIN.getClassName();
    final String url = String.format(pathPrefix + "/connector-plugins/%s/config/validate", plugin);
    final ConfigInfos body = new ConfigInfos(
        plugin,
        1,
        ImmutableList.of("Common"),
        ImmutableList.of(new ConfigInfo(
            new ConfigValueInfo(
                "file",
                ImmutableList.of(
                    "Missing required configuration \"file\" which has no default value."))
        )));

    WireMock.stubFor(
        WireMock.put(WireMock.urlEqualTo(url))
            .withHeader(AUTHORIZATION.toString(), new EqualToPattern(AUTH_HEADER))
            .willReturn(WireMock.aResponse()
                .withStatus(HttpStatus.SC_OK)
                .withBody(MAPPER.writeValueAsString(body)))
    );

    // When:
    final Map<String, String> config = ImmutableMap.of(
        "connector.class", plugin,
        "tasks.max", "1",
        "topics", "test-topic"
    );
    final ConnectResponse<ConfigInfos> response = client.validate(plugin, config);

    // Then:
    assertThat(response.datum(), OptionalMatchers.of(is(body)));
    assertThat("Expected no error!", !response.error().isPresent());
  }