Updated Kafka reader

console/trigger_remove_test.go

@@ -50,4 +50,3 @@ func TestCmdTriggerRemove(t *testing.T) {
 		t.Fatal(err)
 	}
 }
-

data/conf/samples/cdrsexport_internal/cgrates.json

@@ -114,7 +114,7 @@
 				"topic": "cgrates_cdrs",
 			},
 			"tenant": "cgrates.org",
-			"attempts": 1,
+			"attempts": 10,
 			"fields":[
 				{"tag": "RequiredTemplate","type": "*template", "value": "requiredFields"},
 			],

data/conf/samples/cdrsexport_mongo/cgrates.json

@@ -119,7 +119,7 @@
 				"topic": "cgrates_cdrs",
 			},
 			"tenant": "cgrates.org",
-			"attempts": 1,
+			"attempts": 10,
 			"fields":[
 				{"tag": "RequiredTemplate","type": "*template", "value": "requiredFields"},
 			],

data/conf/samples/cdrsexport_mysql/cgrates.json

@@ -115,7 +115,7 @@
 				"topic": "cgrates_cdrs",
 			},
 			"tenant": "cgrates.org",
-			"attempts": 1,
+			"attempts": 10,
 			"fields":[
 				{"tag": "RequiredTemplate","type": "*template", "value": "requiredFields"},
 			],

ers/kafka_it_test.go

@@ -110,7 +110,7 @@ func TestKafkaER(t *testing.T) {
 		if !reflect.DeepEqual(ev.cgrEvent, expected) {
 			t.Errorf("Expected %s ,received %s", utils.ToJSON(expected), utils.ToJSON(ev.cgrEvent))
 		}
-	case <-time.After(10 * time.Second):
+	case <-time.After(30 * time.Second):
 		t.Fatal("Timeout")
 	}
 	close(rdrExit)

general_tests/cdrs_exp_it_test.go

@@ -314,7 +314,7 @@ func testCDRsExpKafka(t *testing.T) {
 
 	defer reader.Close()
 
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	var m kafka.Message
 	var err error
 	if m, err = reader.ReadMessage(ctx); err != nil {

general_tests/cdrs_onlexp_it_test.go

@@ -491,7 +491,7 @@ func testCDRsOnExpKafkaPosterFileFailover(t *testing.T) {
 	defer reader.Close()
 
 	for i := 0; i < 2; i++ { // no raw CDR
-		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 		if m, err := reader.ReadMessage(ctx); err != nil {
 			t.Fatal(err)
 		} else if !reflect.DeepEqual(failoverContent[0], m.Value) && !reflect.DeepEqual(failoverContent[1], m.Value) { // Checking just the prefix should do since some content is dynamic