Improve error handling for triggering function when input data does not conform with input topic schema
@@ -42,6 +42,7 @@ import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SchemaSerializationException;
import org.apache.pulsar.common.functions.FunctionConfig;
import org.apache.pulsar.common.functions.FunctionState;
import org.apache.pulsar.common.functions.Utils;
@@ -1357,6 +1358,8 @@ public abstract class ComponentImpl {
curTime = System.currentTimeMillis();
}
throw new RestException(Status.REQUEST_TIMEOUT, "Request Timed Out");
+ } catch (SchemaSerializationException e) {
+ throw new RestException(Status.BAD_REQUEST, String.format("Failed to serialize input with error: %s. Please check if input data conforms with the schema of the input topic.", e.getMessage()));
} catch (IOException e) {
throw new RestException(Status.INTERNAL_SERVER_ERROR, e.getMessage());
} finally {
