diff --git a/ksqldb-common/src/main/java/io/confluent/ksql/properties/DenyListPropertyValidator.java b/ksqldb-common/src/main/java/io/confluent/ksql/properties/DenyListPropertyValidator.java
index a623955bb2a6..9a2b75cf417b 100644
--- a/ksqldb-common/src/main/java/io/confluent/ksql/properties/DenyListPropertyValidator.java
+++ b/ksqldb-common/src/main/java/io/confluent/ksql/properties/DenyListPropertyValidator.java
@@ -15,18 +15,24 @@
 
 package io.confluent.ksql.properties;
 
+import com.google.common.collect.ImmutableSet;
 import io.confluent.ksql.util.KsqlException;
 
 import java.util.Collection;
 import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
 
 /**
  * Class that validates if a property, or list of properties, is part of a list of denied
  * properties.
  */
-public class DenyListPropertyValidator extends LocalPropertyValidator {
+public class DenyListPropertyValidator {
+  private final Set<String> immutableProps;
+
   public DenyListPropertyValidator(final Collection<String> immutableProps) {
-    super(immutableProps);
+    this.immutableProps = ImmutableSet.copyOf(
+        Objects.requireNonNull(immutableProps, "immutableProps"));
   }
 
   /**
@@ -34,11 +40,10 @@ public DenyListPropertyValidator(final Collection<String> immutableProps) {
    * @throws if a property is part of the denied list.
    */
   public void validateAll(final Map<String, Object> properties) {
-    properties.forEach((k ,v) -> {
-      try {
-        validate(k, v);
-      } catch (final Exception e) {
-        throw new KsqlException(e.getMessage());
+    properties.forEach((name ,v) -> {
+      if (immutableProps.contains(name)) {
+        throw new KsqlException(String.format("A property override was set locally for a "
+            + "property that the server prohibits overrides for: '%s'", name));
       }
     });
   }
diff --git a/ksqldb-common/src/test/java/io/confluent/ksql/properties/DenyListPropertyValidatorTest.java b/ksqldb-common/src/test/java/io/confluent/ksql/properties/DenyListPropertyValidatorTest.java
index 243e45ffa861..f8cf8171d545 100644
--- a/ksqldb-common/src/test/java/io/confluent/ksql/properties/DenyListPropertyValidatorTest.java
+++ b/ksqldb-common/src/test/java/io/confluent/ksql/properties/DenyListPropertyValidatorTest.java
@@ -48,8 +48,10 @@ public void shouldThrowOnDenyListedProperty() {
 
     // Then:
     assertThat(e.getMessage(), containsString(
-        "Cannot override property 'immutable-property'"
+        "A property override was set locally for a property that the server prohibits "
+            + "overrides for: 'immutable-property'"
     ));
+
   }
 
   @Test
diff --git a/ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java b/ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java
index 888c0c42ec95..1697ec24a5f7 100644
--- a/ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java
+++ b/ksqldb-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java
@@ -212,7 +212,7 @@ public EndpointResponse terminateCluster(
           securityContext,
           TERMINATE_CLUSTER,
           new SessionProperties(
-              request.getStreamsProperties(),
+              streamsProperties,
               localHost,
               localUrl,
               false
diff --git a/ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/KsqlResourceTest.java b/ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/KsqlResourceTest.java
index c87db6ba7d33..bff57af6d017 100644
--- a/ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/KsqlResourceTest.java
+++ b/ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/KsqlResourceTest.java
@@ -2214,7 +2214,8 @@ public void shouldThrowOnDenyListedStreamProperty() {
     // Then:
     assertThat(response.getStatus(), CoreMatchers.is(BAD_REQUEST.code()));
     assertThat(((KsqlErrorMessage) response.getEntity()).getMessage(),
-        is("Cannot override property '" + StreamsConfig.NUM_STREAM_THREADS_CONFIG + "'"));
+        is("A property override was set locally for a property that the server prohibits "
+            + "overrides for: '" + StreamsConfig.NUM_STREAM_THREADS_CONFIG + "'"));
   }
 
   @Test
@@ -2251,7 +2252,8 @@ public void shouldThrowOnDenyListedConfigProperty() {
     // Then:
     assertThat(response.getStatus(), CoreMatchers.is(BAD_REQUEST.code()));
     assertThat(((KsqlErrorMessage) response.getEntity()).getMessage(),
-        is("Cannot override property '" + StreamsConfig.NUM_STREAM_THREADS_CONFIG + "'"));
+        is("A property override was set locally for a property that the server prohibits "
+            + "overrides for: '" + StreamsConfig.NUM_STREAM_THREADS_CONFIG + "'"));
   }
 
   private void givenKsqlConfigWith(final Map<String, Object> additionalConfig) {
diff --git a/ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResourceTest.java b/ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResourceTest.java
index 9c15c33e2dca..66173feaf487 100644
--- a/ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResourceTest.java
+++ b/ksqldb-rest-app/src/test/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResourceTest.java
@@ -382,7 +382,8 @@ public void shouldThrowOnDenyListedStreamProperty() {
         StreamsConfig.NUM_STREAM_THREADS_CONFIG);
     testResource.configure(new KsqlConfig(props));
     when(errorsHandler.generateResponse(any(), any()))
-        .thenReturn(badRequest("Cannot override property 'num.stream.threads'"));
+        .thenReturn(badRequest("A property override was set locally for a property that the "
+            + "server prohibits overrides for: 'num.stream.threads'"));
 
     // When:
     final EndpointResponse response = testResource.streamQuery(
@@ -400,7 +401,8 @@ public void shouldThrowOnDenyListedStreamProperty() {
     // Then:
     assertThat(response.getStatus(), CoreMatchers.is(BAD_REQUEST.code()));
     assertThat(((KsqlErrorMessage) response.getEntity()).getMessage(),
-        is("Cannot override property '" + StreamsConfig.NUM_STREAM_THREADS_CONFIG + "'"));
+        is("A property override was set locally for a property that the server prohibits "
+            + "overrides for: '" + StreamsConfig.NUM_STREAM_THREADS_CONFIG + "'"));
   }
 
   @Test