public EndpointResponse isValidProperty(final String property) {
    try {
      final Map<String, Object> properties = new HashMap<>();
      properties.put(property, "");
      denyListPropertyValidator.validateAll(properties);
      final KsqlConfigResolver resolver = new KsqlConfigResolver();
      final Optional<ConfigItem> resolvedItem = resolver.resolve(property, false);
      if (ksqlEngine.getKsqlConfig().getBoolean(KsqlConfig.KSQL_SHARED_RUNTIME_ENABLED)
          && resolvedItem.isPresent()) {
        if (!PropertiesList.QueryLevelProperties.contains(resolvedItem.get().getPropertyName())) {
          throw new KsqlException(String.format("When shared runtimes are enabled, the"
              + " config %s can only be set for the entire cluster and all queries currently"
              + " running in it, and not configurable for individual queries."
              + " Please use ALTER SYSTEM to change this config for all queries.",
              properties));
        }
      }
      return EndpointResponse.ok(true);
    } catch (final KsqlException e) {
      LOG.info("Processed unsuccessfully, reason: ", e);
      return errorHandler.generateResponse(e, Errors.badRequest(e));
    } catch (final Exception e) {
      LOG.info("Processed unsuccessfully, reason: ", e);
      throw e;
    }
  }
@Test
  public void shouldReturnBadRequestWhenIsValidatorIsCalledWithProhibitedProps() {
    final Map<String, Object> properties = new HashMap<>();
    properties.put("ksql.service.id", "");

    // Given:
    doThrow(new KsqlException("deny override")).when(denyListPropertyValidator).validateAll(
        properties
    );

    // When:
    final EndpointResponse response = ksqlResource.isValidProperty("ksql.service.id");

    // Then:
    assertThat(response.getStatus(), equalTo(400));
  }