Improved/fixed aggregate function error messages.

ksqldb-engine/src/main/java/io/confluent/ksql/function/udaf/max/MaxAggFunctionFactory.java

@@ -54,9 +54,13 @@ public KsqlAggregateFunction createAggregateFunction(
       case TIMESTAMP:
         return new MaxKudaf(FUNCTION_NAME, initArgs.udafIndex(), argSchema);
       default:
-        throw new KsqlException("No MAX aggregate function with " + argTypeList.get(0) + " "
-            + "argument type exists!");
-
+        throw new KsqlException(
+          String.format(
+            "No %s aggregate function with %s argument type exists!",
+            FUNCTION_NAME,
+            argTypeList.get(0)
+          )
+        );
     }
   }
 

ksqldb-engine/src/main/java/io/confluent/ksql/function/udaf/min/MinAggFunctionFactory.java

@@ -53,9 +53,13 @@ public KsqlAggregateFunction createAggregateFunction(
       case TIMESTAMP:
         return new MinKudaf(FUNCTION_NAME, initArgs.udafIndex(), argSchema);
       default:
-        throw new KsqlException("No MIN aggregate function with " + argTypeList.get(0) + " "
-            + "argument type exists!");
-
+        throw new KsqlException(
+          String.format(
+            "No %s aggregate function with %s argument type exists!",
+            FUNCTION_NAME,
+            argTypeList.get(0)
+          )
+        );
     }
   }
 

ksqldb-engine/src/main/java/io/confluent/ksql/function/udaf/sum/SumAggFunctionFactory.java

@@ -55,9 +55,13 @@ public KsqlAggregateFunction createAggregateFunction(
       case DECIMAL:
         return new DecimalSumKudaf(FUNCTION_NAME, initArgs.udafIndex(), (SqlDecimal) argSchema);
       default:
-        throw new KsqlException("No Max aggregate function with " + argTypeList.get(0) + " "
-            + " argument type exists!");
-
+        throw new KsqlException(
+          String.format(
+            "No %s aggregate function with %s argument type exists!",
+            FUNCTION_NAME,
+            argTypeList.get(0)
+          )
+        );
     }
   }
 

ksqldb-engine/src/main/java/io/confluent/ksql/function/udaf/topk/TopKAggregateFunctionFactory.java

@@ -92,8 +92,13 @@ public KsqlAggregateFunction createAggregateFunction(
             Collections.singletonList(ParamTypes.STRING),
             String.class);
       default:
-        throw new KsqlException("No TOPK aggregate function with " + argumentType.get(0)
-            + " argument type exists!");
+        throw new KsqlException(
+          String.format(
+            "No %s aggregate function with %s argument type exists!",
+            NAME,
+            argumentType.get(0)
+          )
+        );
     }
   }
 

ksqldb-engine/src/main/java/io/confluent/ksql/function/udaf/topkdistinct/TopkDistinctAggFunctionFactory.java

@@ -69,8 +69,13 @@ public KsqlAggregateFunction createAggregateFunction(
             SchemaConverters.sqlToFunctionConverter().toFunctionType(argSchema),
             SchemaConverters.sqlToJavaConverter().toJavaType(argSchema));
       default:
-        throw new KsqlException("No TOPKDISTINCT aggregate function with " + argTypeList.get(0)
-            + " argument type exists!");
+        throw new KsqlException(
+          String.format(
+            "No %s aggregate function with %s argument type exists!",
+            NAME,
+            argTypeList.get(0)
+          )
+        );
     }
   }
 

ksqldb-engine/src/test/java/io/confluent/ksql/engine/KsqlEngineTest.java

@@ -310,6 +310,25 @@ public void shouldThrowWhenExecutingInsertIntoTable() {
     assertThat(e, statementText(is("insert into bar select * from test2;")));
   }
 
+  @Test
+  public void shouldThrowForBadSumAggregate() {
+    // When:
+    final KsqlStatementException e = assertThrows(
+        KsqlStatementException.class,
+        () -> KsqlEngineTestUtil.executeQuery(
+            serviceContext,
+            ksqlEngine,
+            "SELECT 'dummy', SUM(CAST(col1 AS STRING)) FROM test1 GROUP BY 'dummy' EMIT CHANGES;",
+            ksqlConfig,
+            Collections.emptyMap()
+        )
+    );
+
+    // Then:
+    assertThat(e, rawMessage(containsString(
+        "No SUM aggregate function with STRING argument type exists!")));
+  }
+
   @Test
   public void shouldThrowOnInsertIntoStreamWithTableResult() {
     KsqlEngineTestUtil.execute(