Fix InputTopic Submission for Creating Sink (apache#2427)

* Fix Spelling mistake in error message * Fix InputTopic Submission for Creating Sink
vruc · Aug 23, 2018 · 4694e6e · 4694e6e
1 parent 001445c
commit 4694e6e
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/...ions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java b/...ions/utils/src/main/java/org/apache/pulsar/functions/utils/validation/ValidatorImpls.java
@@ -547,7 +547,7 @@ private static void doCommonChecks(FunctionConfig functionConfig) {
             if (functionConfig.getTimeoutMs() != null
                     && functionConfig.getProcessingGuarantees() != null
                     && functionConfig.getProcessingGuarantees() != FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE) {
-                throw new IllegalArgumentException("Message timeout can only be specifed with processing guarantee is "
+                throw new IllegalArgumentException("Message timeout can only be specified with processing guarantee is "
                         + FunctionConfig.ProcessingGuarantees.ATLEAST_ONCE.name());
             }
         }
@@ -817,6 +817,9 @@ public void validateField(String name, Object o) {
 
         private static Collection<String> collectAllInputTopics(SinkConfig sinkConfig) {
             List<String> retval = new LinkedList<>();
+            if (sinkConfig.getInputs() != null) {
+                retval.addAll(sinkConfig.getInputs());
+            }
             if (sinkConfig.getTopicToSerdeClassName() != null) {
                 retval.addAll(sinkConfig.getTopicToSerdeClassName().keySet());
             }