chore: re-raise UDFError instead of dropping messages

[KeranYang]

closes #68

Manual E2E Testing Notes:

After this change, if UDF throws an exception, the exception immediately get re-raised and Numaflow will keep retrying on failed messages until the exception gets resolved. In my test, since the exception never gets resolved, the retry keeps going on forever.

UDF numa log

ERROR 2023-03-30 15:03:46 {"level":"error","ts":1680203026.7438965,"logger":"numaflow.MapUDF-processor","caller":"forward/forward.go:561","msg":"UDF.Apply error","vertex":"err-test-throw-error","protocol":"uds-grpc-map-udf","error":"gRPC client.MapFn failed, failed to execute c.grpcClt.MapFn(): rpc error: code = Unknown desc = Exception calling application: Sorry, raising an exception","stacktrace":"github.com/numaproj/numaflow/pkg/forward.(*InterStepDataForward).applyUDF\n\t/Users/kyang5/Desktop/development/numaflow/numaflow/pkg/forward/forward.go:561\ngithub.com/numaproj/numaflow/pkg/forward.(*InterStepDataForward).concurrentApplyUDF\n\t/Users/kyang5/Desktop/development/numaflow/numaflow/pkg/forward/forward.go:546\ngithub.com/numaproj/numaflow/pkg/forward.(*InterStepDataForward).forwardAChunk.func1\n\t/Users/kyang5/Desktop/development/numaflow/numaflow/pkg/forward/forward.go:264"}
ERROR 2023-03-30 15:03:46 {"level":"error","ts":1680203026.7438965,"logger":"numaflow.MapUDF-processor","caller":"forward/forward.go:561","msg":"UDF.Apply error","vertex":"err-test-throw-error","protocol":"uds-grpc-map-udf","error":"gRPC client.MapFn failed, failed to execute c.grpcClt.MapFn(): rpc error: code = Unknown desc = Exception calling application: Sorry, raising an exception","stacktrace":"github.com/numaproj/numaflow/pkg/forward.(*InterStepDataForward).applyUDF\n\t/Users/kyang5/Desktop/development/numaflow/numaflow/pkg/forward/forward.go:561\ngithub.com/numaproj/numaflow/pkg/forward.(*InterStepDataForward).concurrentApplyUDF\n\t/Users/kyang5/Desktop/development/numaflow/numaflow/pkg/forward/forward.go:546\ngithub.com/numaproj/numaflow/pkg/forward.(*InterStepDataForward).forwardAChunk.func1\n\t/Users/kyang5/Desktop/development/numaflow/numaflow/pkg/forward/forward.go:264"}
...
...

UDF udf log

2023-03-30 19:04:18 CRITICAL UDFError, re-raising the error: Exception('Sorry, raising an exception')
Exception: Sorry, raising an exception
raise Exception("Sorry, raising an exception")
File "/app/example.py", line 10, in my_handler
msgs = self.__map_handler(
File "/opt/pysetup/.venv/lib/python3.10/site-packages/pynumaflow/function/server.py", line 135, in MapFn
Traceback (most recent call last):
2023-03-30 19:04:18 CRITICAL UDFError, re-raising the error: Exception('Sorry, raising an exception')
Exception: Sorry, raising an exception
raise Exception("Sorry, raising an exception")
File "/app/example.py", line 10, in my_handler
msgs = self.__map_handler(
File "/opt/pysetup/.venv/lib/python3.10/site-packages/pynumaflow/function/server.py", line 135, in MapFn
Traceback (most recent call last):
...
...

I sent two messages using HTTP. Messages stay in Pending status.

[codecov[bot]]

Codecov Report

Merging #71 (72f9a97) into main (6fa43d6) will increase coverage by 1.13%. The diff coverage is 100.00%.

@@            Coverage Diff             @@
##             main      #71      +/-   ##
==========================================
+ Coverage   97.03%   98.17%   +1.13%     
==========================================
  Files          11       11              
  Lines         439      439              
  Branches       34       34              
==========================================
+ Hits          426      431       +5     
+ Misses          8        4       -4     
+ Partials        5        4       -1     

Impacted Files	Coverage Δ
pynumaflow/function/server.py	97.58% <100.00%> (+4.03%)	:arrow_up:

:mega: We’re building smart automated test selection to slash your CI/CD build times. Learn more