Repository: argoproj-labs/old-argo-dataflow (language: python)

Issue:
Dead Letter Queues failing 
_cross-posting_

> Hey,
I see that v0.3.0 has support for Dead letter queues. I just gave it a spin as per [this comment from the PR](https://github.com/argoproj-labs/argo-dataflow/pull/444#issuecomment-938754302) - but got stuck at a few places

> **Context:** I have a kafka source and a sink for the first step in my pipeline and the messages from the sink will be consumed by the subsequent step.


> **Problem:** I wanted to know if the failed messages in the first step can be replayed through a DLQ? if so how do I add the failed messages to a DLQ? (I tried adding another kafka sink and marking it as a DLQ, didn't work as expected)

> If I keep the same name for both sinks, it errs out.


<img width="473" alt="Screenshot 2021-10-28 at 12 59 56 AM" src="https://user-images.githubusercontent.com/20226361/139207501-b382189f-adcf-459f-b707-2478a89146b6.png">

<img width="573" alt="Screenshot 2021-10-28 at 1 00 07 AM" src="https://user-images.githubusercontent.com/20226361/139207510-09e93c49-91bb-4199-91e9-f61c8ec663f4.png">


Discussion / hints:
OK, you need to give the DLQ a name that is not the same other sink.
@inishchith All sink name should be unique. Give new name like `DLQ` and try
Initially, I did try with different sink names, it threw the following error (image) and reached out on the channel.

> did you give different names to the DLQ and othor sink? if, not this is a bug, could you please raise it?
> by @alexec 

![Screenshot 2021-10-27 at 6 07 04 PM](https://user-images.githubusercontent.com/20226361/139293366-497fb231-e3b7-4f6c-9eda-0cc9992e7af4.png)

Think I might have misunderstood something, LMK. Thanks!





@inishchith Can you provide the sample pipeline for the above issue to reproduce? 
@sarabala1979 here.

[test_pipeline.txt](https://github.com/argoproj-labs/argo-dataflow/files/7436238/test_pipeline.txt)

@inishchith  I found the bug. working on fix.

Fix patch:
diff --git a/runner/sidecar/sink/kafka/kafka.go b/runner/sidecar/sink/kafka/kafka.go
index 07e46c48..e0af749f 100644
--- a/runner/sidecar/sink/kafka/kafka.go
+++ b/runner/sidecar/sink/kafka/kafka.go
@@ -26,6 +26,22 @@ type kafkaSink struct {
 	async    bool
 }
 
+var kafkaMessagesProducedSuccess, kafkaMessagesProducedErr *prometheus.CounterVec
+
+func init() {
+	// track async success and errors
+	kafkaMessagesProducedSuccess = promauto.NewCounterVec(prometheus.CounterOpts{
+		Subsystem: "sinks",
+		Name:      "kafka_produced_successes",
+		Help:      "Number of messages successfully produced to Kafka",
+	}, []string{"sinkName"})
+	kafkaMessagesProducedErr = promauto.NewCounterVec(prometheus.CounterOpts{
+		Subsystem: "sinks",
+		Name:      "kafka_produce_errors",
+		Help:      "Number of errors while producing messages to Kafka",
+	}, []string{"sinkName"})
+}
+
 func New(ctx context.Context, sinkName string, secretInterface corev1.SecretInterface, x dfv1.KafkaSink) (sink.Interface, error) {
 	logger := logger.WithValues("sink", sinkName)
 	config, err := sharedkafka.GetConfig(ctx, secretInterface, x.KafkaConfig)
@@ -50,7 +66,6 @@ func New(ctx context.Context, sinkName string, secretInterface corev1.SecretInte
 	if err != nil {
 		return nil, err
 	}
-
 	go wait.JitterUntilWithContext(ctx, func(context.Context) {
 		logger.Info("consuming Kafka logs")
 		for e := range producer.Logs() {
@@ -58,18 +73,6 @@ func New(ctx context.Context, sinkName string, secretInterface corev1.SecretInte
 		}
 	}, 3*time.Second, 1.2, true)
 
-	// track async success and errors
-	kafkaMessagesProducedSuccess := promauto.NewCounterVec(prometheus.CounterOpts{
-		Subsystem: "sinks",
-		Name:      "kafka_produced_successes",
-		Help:      "Number of messages successfully produced to Kafka",
-	}, []string{"sinkName"})
-	kafkaMessagesProducedErr := promauto.NewCounterVec(prometheus.CounterOpts{
-		Subsystem: "sinks",
-		Name:      "kafka_produce_errors",
-		Help:      "Number of errors while producing messages to Kafka",
-	}, []string{"sinkName"})
-
 	go wait.JitterUntilWithContext(ctx, func(context.Context) {
 		logger.Info("starting producer event consuming loop")
 		for e := range producer.Events() {
diff --git a/runner/sidecar/sinks.go b/runner/sidecar/sinks.go
index a481ec04..c2174c5e 100644
--- a/runner/sidecar/sinks.go
+++ b/runner/sidecar/sinks.go
@@ -32,6 +32,7 @@ func connectSinks(ctx context.Context) (func(context.Context, []byte) error, fun
 		Name:      "errors",
 		Help:      "Total number of errors, see https://github.com/argoproj-labs/argo-dataflow/blob/main/docs/METRICS.md#sinks_errors",
 	}, []string{"sinkName", "replica", "dlq"})
+
 	for _, s := range step.Spec.Sinks {
 		logger.Info("connecting sink", "sink", sharedutil.MustJSON(s))
 		sinkName := s.Name


Test patch:
diff --git a/test/http-fmea/http_fmea_test.go b/test/http-fmea/http_fmea_test.go
index ab057018..4f4e0b8d 100644
--- a/test/http-fmea/http_fmea_test.go
+++ b/test/http-fmea/http_fmea_test.go
@@ -89,3 +89,38 @@ func TestHTTPFMEA_PodDeletedDisruption_TwoReplicas(t *testing.T) {
 	WaitForCounter(2*n, 2*n)
 	WaitForNoErrors()
 }
+
+func TestHTTPFMEA_PodDeletedDisruption_OneReplica(t *testing.T) {
+	defer Setup(t)()
+
+	CreatePipeline(Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: "http"},
+		Spec: PipelineSpec{
+			Steps: []StepSpec{{
+				Name:    "main",
+				Cat:     &Cat{},
+				Sources: []Source{{HTTP: &HTTPSource{}}},
+				Sinks:   []Sink{{HTTP: &HTTPSink{URL: "http://testapi/count/incr"}}},
+			}},
+		},
+	})
+
+	WaitForPipeline()
+
+	n := 15
+
+	// with a single replica, if you loose a replica, you loose service
+	go PumpHTTPTolerantly(n)
+
+	WaitForPod("http-main-0")
+	stopPortForward := StartPortForward("http-main-0")
+	WaitForSunkMessages()
+	stopPortForward()
+
+	DeletePod("http-main-0") // delete the pod to see that we recover and continue to process messages
+	WaitForPod("http-main-0")
+
+	WaitForCounter(n, n)
+	defer StartPortForward("http-main-0")()
+	WaitForNoErrors()
+}
\ No newline at end of file
diff --git a/test/kafka-e2e/kafka_test.go b/test/kafka-e2e/kafka_test.go
index 67178f9e..101e3162 100644
--- a/test/kafka-e2e/kafka_test.go
+++ b/test/kafka-e2e/kafka_test.go
@@ -82,3 +82,40 @@ func TestKafkaAsync(t *testing.T) {
 	DeletePipelines()
 	WaitForPodsToBeDeleted()
 }
+
+func TestKafkaMultipleSink(t *testing.T) {
+	defer Setup(t)()
+
+	topic := CreateKafkaTopic()
+	sinkTopic := CreateKafkaTopic()
+	sinkTopic1 := CreateKafkaTopic()
+
+	CreatePipeline(Pipeline{
+		ObjectMeta: metav1.ObjectMeta{GenerateName: "kafka-"},
+		Spec: PipelineSpec{
+			Steps: []StepSpec{{
+				Name:    "main",
+				Cat:     &Cat{},
+				Sources: []Source{{Kafka: &KafkaSource{Kafka: Kafka{Topic: topic}}}},
+				Sinks: []Sink{{Kafka: &KafkaSink{Kafka: Kafka{Topic: sinkTopic}, Async: true}},
+					{Name: "sink2", Kafka: &KafkaSink{Kafka: Kafka{Topic: sinkTopic1}, Async: true}}},
+			}},
+		},
+	})
+	WaitForPipeline()
+	WaitForPod()
+
+	defer StartPortForward()()
+
+	PumpKafkaTopic(topic, 17)
+
+	WaitForSunkMessages()
+
+	WaitForTotalSourceMessages(17)
+	WaitForTotalSunkMessages(17)
+
+	ExpectMetric("sinks_kafka_produced_successes", Eq(17))
+
+	DeletePipelines()
+	WaitForPodsToBeDeleted()
+}