chore: fix merge conflicts

validator/cmd/centralconsumer/main.go

@@ -20,7 +20,7 @@ import (
 )
 
 func main() {
-	configFile := flag.String("f", "", "toml file containing configuration of working environment")
+	configFile := flag.String("f", "./config/validator.toml", "toml file containing configuration of working environment")
 	flag.Parse()
 
 	janitorctl.RunCentralConsumer(*configFile)

validator/config/validator.toml

@@ -18,14 +18,19 @@ schema_id = ""
 schema_version = ""
 schema_type = ""
 
+validate_header = "true"
+
+#default_header_schema_id = "5"
+#default_header_schema_version = "1"
+
 [consumer]
-type = "" # insert "kafka", "pubsub", "servicebus" or "jetstream"
+type = "kafka" # insert "kafka", "pubsub", "servicebus" or "jetstream"
 encryption_key = ""
 
 [consumer.kafka]
-address = "" # insert
-topic = "" # insert
-group_id = "" # insert
+address = "34.116.158.107:9094" # insert
+topic = "input-topic" # insert
+group_id = "Group" # insert
 
 [consumer.pubsub]
 project_id = "" # insert
@@ -42,10 +47,13 @@ subject = "" # insert
 consumer_name = "" # insert
 
 [producer]
-type = "" # insert "kafka", "eventhubs", "pubsub", "servicebus" or "jetstream"
+type = "kafka" # insert "kafka", "eventhubs", "pubsub", "servicebus" or "jetstream"
 
 [producer.kafka]
-address = "" # insert
+address = "34.116.158.107:9094" # insert
+
+[producer.kafka.settings]
+kafka_acks = "1"
 
 [producer.pubsub]
 project_id = "" # insert
@@ -57,18 +65,31 @@ connection_string = "" # insert
 url = "" # insert
 
 [topics]
-valid = "" # insert
-dead_letter = "" # insert
+valid = "valid-topic" # insert
+dead_letter = "dead-letter-topic" # insert
 
 [registry]
-url = ""
+url = "http://34.118.10.21:8080"
 type = "janitor" # insert "janitor" or "apicurio"
-groupID = "default"
+groupID = "group"
 
 [validators]
 enable_json = "true"
+enable_avro = "true"
 enable_csv = "false"
 enable_xml = "false"
 enable_protobuf = "false"
-csv_url = "http://csv-validator-svc:8080"
+csv_url = "http://localhost:8081"
 xml_url = "http://xml-validator-svc:8081"
+
+[should_log]
+missing_schema = "true"
+valid = "true"
+dead_letter = "true"
+
+[default_header_schema]
+id = "9"
+version = "1"
+
+[run_option]
+err_threshold = 100

validator/internal/centralconsumer/centralconsumer.go

@@ -97,8 +97,8 @@ type Settings struct {
 	// NumInferrers defines the maximum amount of inflight destination topic inference jobs (validation and routing).
 	NumInferrers int
 
-	// ValidateHeaders defines if the messages' headers will be validated
-	ValidateHeaders bool
+	// ValidateHeader defines if the messages' headers will be validated
+	ValidateHeader bool
 
 	// DefaultHeaderSchemaId is default ID of the header schema
 	DefaultHeaderSchemaId string
@@ -143,7 +143,7 @@ func New(registry registry.SchemaRegistry, publisher broker.Publisher, validator
 	if settings.NumInferrers > 0 {
 		validatorsSem = make(chan struct{}, settings.NumInferrers)
 	}
-	if settings.ValidateHeaders {
+	if settings.ValidateHeader {
 		_, ok := validators["json"]
 		if !ok {
 			// if json validation is turned off, this version of json validator is used by default for validating message header
@@ -200,7 +200,7 @@ func New(registry registry.SchemaRegistry, publisher broker.Publisher, validator
 			Specification: schemaVersion.Specification,
 		},
 		encryptionKey:   encryptionKey,
-		validateHeaders: settings.ValidateHeaders,
+		validateHeaders: settings.ValidateHeader,
 		defaultHeaderSchema: config.DefaultHeaderSchema{
 			DefaultHeaderSchemaId:      settings.DefaultHeaderSchemaId,
 			DefaultHeaderSchemaVersion: settings.DefaultHeaderSchemaVersion,

validator/internal/config/centralconsumer.go

@@ -30,7 +30,7 @@ type CentralConsumer struct {
 	ShouldLog              CentralConsumerShouldLog  `toml:"should_log"`
 	NumSchemaCollectors    int                       `toml:"num_schema_collectors" default:"-1"`
 	NumInferrers           int                       `toml:"num_inferrers" default:"-1"`
-	ValidateHeaders        bool                      `toml:"validate_headers"`
+	ValidateHeader         bool                      `toml:"validate_header"`
 	DefaultHeaderSchema    DefaultHeaderSchema       `toml:"default_header_schema"`
 	MetricsLoggingInterval time.Duration             `toml:"metrics_logging_interval" default:"5s"`
 	RunOptions             RunOptions                `toml:"run_option"`

validator/internal/janitorctl/run.go

@@ -86,7 +86,7 @@ func RunCentralConsumer(configFile string) {
 			centralconsumer.Settings{
 				NumSchemaCollectors:        cfg.NumSchemaCollectors,
 				NumInferrers:               cfg.NumInferrers,
-				ValidateHeaders:            cfg.ValidateHeaders,
+				ValidateHeader:             cfg.ValidateHeader,
 				DefaultHeaderSchemaId:      cfg.DefaultHeaderSchema.DefaultHeaderSchemaId,
 				DefaultHeaderSchemaVersion: cfg.DefaultHeaderSchema.DefaultHeaderSchemaVersion,
 			},