feat: add kafka rack to consumer configurations

Author: Abdulsametileri

README.md

@@ -87,6 +87,7 @@ func consumeFn(message kafka.Message) error {
 | `logLevel`                         | Describes log level; valid options are `debug`, `info`, `warn`, and `error`                                                           | info             |                          |
 | `concurrency`                      | Number of goroutines used at listeners                                                                                                | runtime.NumCPU() |
 | `retryEnabled`                     | Retry/Exception consumer is working or not                                                                                            | false            |
+| `rack`                             | [see doc](https://pkg.go.dev/github.com/segmentio/kafka-go#RackAffinityGroupBalancer)                                                 |                  |
 | `retryConfiguration.startTimeCron` | Cron expression when retry consumer ([kafka-cronsumer](https://github.com/Trendyol/kafka-cronsumer#configurations)) starts to work at |                  |
 | `retryConfiguration.workDuration`  | Work duration exception consumer actively consuming messages                                                                          |                  |
 | `retryConfiguration.topic`         | Retry/Exception topic names                                                                                                           |                  |

consumer_config.go

@@ -14,6 +14,7 @@ type ConsumeFn func(Message) error
 type ConsumerConfig struct {
 	Reader ReaderConfig
 
+	Rack string
 	SASL *SASLConfig
 	TLS  *TLSConfig
 
@@ -61,6 +62,9 @@ func (c *ConsumerConfig) newKafkaReader() (*kafka.Reader, error) {
 
 	reader := kafka.ReaderConfig(c.Reader)
 	reader.Dialer = dialer
+	if c.Rack != "" {
+		reader.GroupBalancers = []kafka.GroupBalancer{kafka.RackAffinityGroupBalancer{Rack: c.Rack}}
+	}
 
 	return kafka.NewReader(reader), nil
 }