fix: error handling in case of udsink failures

[adarsh0728]

@yhl25 and I were debugging a scenario where udsink panics. We observed that numa container wasn't restarting as it should. Fixed the error handling in this case.

Before PR:

2025-01-28T05:21:06.974698Z ERROR numaflow_core::source: Nak received for offset offset=String(StringOffset { offset: b"MTczODA0MTY2Njk0NzEwODU2NC0x", partition_idx: 0 })
2025-01-28T05:21:06.974694Z ERROR numaflow_core::source: Nak received for offset offset=String(StringOffset { offset: b"MTczODA0MTY2Njk0NzEwODU2NC0w", partition_idx: 0 })
2025-01-28T05:21:06.974628Z ERROR numaflow_core::sink: Error writing to sink e=Grpc("status: Unknown, message: \"panic inside sink handler: sink fn panicked on multiple of 3\", details: [], metadata: MetadataMap { headers: {} }")

With this PR:

2025-01-28T06:50:48.818759Z  INFO numaflow_core::metrics: Stopped the Lag-Reader Expose and Builder tasks
2025-01-28T06:50:48.818764Z ERROR numaflow_core: Application error running monovertex: Grpc("status: Unknown, message: \"panic inside sink handler: sink fn panicked on multiple of 3\", details: [], metadata: MetadataMap { headers: {} }")
2025-01-28T06:50:48.818768Z  INFO numaflow_core: Gracefully Exiting...

How to generate?

apiVersion: numaflow.numaproj.io/v1alpha1  
  
kind: MonoVertex  
  
metadata:  
  name: simple-mono-vertex  
spec:  
  source:  
    generator:  
      rpu: 50  
      duration: 1s  
  sink:  
    udsink:  
      container:  
        image: quay.io/adarsh0728/test/sink-log:stable

[codecov]

Codecov Report

Attention: Patch coverage is 75.00000% with 1 line in your changes missing coverage. Please review.

Project coverage is 67.96%. Comparing base (36f8357) to head (0c98071).

Files with missing lines	Patch %	Lines
rust/numaflow-core/src/source.rs	66.66%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2372      +/-   ##
==========================================
- Coverage   68.14%   67.96%   -0.19%     
==========================================
  Files         352      352              
  Lines       46500    46500              
==========================================
- Hits        31688    31604      -84     
- Misses      13732    13815      +83     
- Partials     1080     1081       +1     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[vigith]

were we sending the wrong metrics during error because of this logical bug?

[yhl25]

This is the case when udsink container goes down because of a panic. Numa was not getting restarted immediately instead it was getting restarted after the multiple health check failure.

[adarsh0728]

Encountered during testing of changes I'm making in this PR