Skip to content

Commit 541cca7

Browse files
AbdulsametileriAbdulsametileri
authored
feat: cronsumer internal queue capacity, producer batch size and timeout expose (#147)
* feat: able to use kafka cronsumer queue capacity field * feat: able to inject cronsumer producer batch size and timeout * docs: add new cronsumer exposed field to the readme * feat: kafka cronsumer v1.1.5 dump
1 parent b9c0b2b commit 541cca7

File tree

6 files changed

+24
-9
lines changed

6 files changed

+24
-9
lines changed

‎README.md

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -241,17 +241,20 @@ under [the specified folder](examples/with-sasl-plaintext) and then start the ap
241241
| `metricPrefix` | MetricPrefix is used for prometheus fq name prefix. If not provided, default metric prefix value is `kafka_konsumer`. Currently, there are two exposed prometheus metrics. `processed_messages_total` and `unprocessed_messages_total` So, if default metric prefix used, metrics names are `kafka_konsumer_processed_messages_total_current` and `kafka_konsumer_unprocessed_messages_total_current`. | kafka_konsumer |
242242
| `dial.Timeout` | [see doc](https://pkg.go.dev/github.com/segmentio/kafka-go#Dialer) | no timeout |
243243
| `dial.KeepAlive` | [see doc](https://pkg.go.dev/github.com/segmentio/kafka-go#Dialer) | not enabled |
244-
| `transport.DialTimeout ` | [see doc](https://pkg.go.dev/github.com/segmentio/kafka-go@#Transport) | 5s |
245-
| `transport.IdleTimeout ` | [see doc](https://pkg.go.dev/github.com/segmentio/kafka-go@#Transport) | 30s |
246-
| `transport.MetadataTTL ` | [see doc](https://pkg.go.dev/github.com/segmentio/kafka-go@#Transport) | 6s |
247-
| `transport.MetadataTopics ` | [see doc](https://pkg.go.dev/github.com/segmentio/kafka-go@#Transport) | all topics in cluster |
244+
| `transport.DialTimeout ` | [see doc]((https://pkg.go.dev/github.com/segmentio/kafka-go@v0.4.47#Transport) | 5s |
245+
| `transport.IdleTimeout ` | [see doc]((https://pkg.go.dev/github.com/segmentio/kafka-go@v0.4.47#Transport) | 30s |
246+
| `transport.MetadataTTL ` | [see doc]((https://pkg.go.dev/github.com/segmentio/kafka-go@v0.4.47#Transport) | 6s |
247+
| `transport.MetadataTopics ` | [see doc]((https://pkg.go.dev/github.com/segmentio/kafka-go@v0.4.47#Transport) | all topics in cluster |
248248
| `distributedTracingEnabled` | indicates open telemetry support on/off for consume and produce operations. | false |
249249
| `distributedTracingConfiguration.TracerProvider` | [see doc](https://opentelemetry.io/docs/specs/otel/trace/api/) | otel.GetTracerProvider() |
250250
| `distributedTracingConfiguration.Propagator` | [see doc](https://opentelemetry.io/docs/specs/otel/trace/api/) | otel.GetTextMapPropagator() |
251-
| `retryConfiguration.clientId` | [see doc](https://pkg.go.dev/github.com/segmentio/kafka-go@#Transport) | |
251+
| `retryConfiguration.clientId` | [see doc]((https://pkg.go.dev/github.com/segmentio/kafka-go@v0.4.47#Transport) | |
252252
| `retryConfiguration.startTimeCron` | Cron expression when retry consumer ([kafka-cronsumer](https://github.com/Trendyol/kafka-cronsumer#configurations)) starts to work at | |
253253
| `retryConfiguration.metricPrefix` | MetricPrefix is used for prometheus fq name prefix. If not provided, default metric prefix value is kafka_cronsumer. Currently, there are two exposed prometheus metrics. retried_messages_total_current and discarded_messages_total_current. So, if default metric prefix used, metrics names are kafka_cronsumer_retried_messages_total_current and kafka_cronsumer_discarded_messages_total_current | kafka_cronsumer |
254254
| `retryConfiguration.workDuration` | Work duration exception consumer actively consuming messages | |
255+
| `retryConfiguration.queueCapacity` | [see doc](https://pkg.go.dev/github.com/segmentio/[email protected]#ReaderConfig.QueueCapacity) | 100 |
256+
| `retryConfiguration.producerBatchSize` | [see doc](https://pkg.go.dev/github.com/segmentio/[email protected]#WriterConfig.BatchSize) | 100 |
257+
| `retryConfiguration.producerBatchTimeout` | [see doc](https://pkg.go.dev/github.com/segmentio/[email protected]#WriterConfig.BatchTimeout) | 100 |
255258
| `retryConfiguration.topic` | Retry/Exception topic names | |
256259
| `retryConfiguration.brokers` | Retry topic brokers urls | |
257260
| `retryConfiguration.maxRetry` | Maximum retry value for attempting to retry a message | 3 |

‎consumer_config.go

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -126,6 +126,7 @@ func (cfg *ConsumerConfig) newCronsumerConfig() *kcronsumer.Config {
126126
MaxRetry: cfg.RetryConfiguration.MaxRetry,
127127
VerifyTopicOnStartup: cfg.RetryConfiguration.VerifyTopicOnStartup,
128128
Concurrency: cfg.RetryConfiguration.Concurrency,
129+
QueueCapacity: cfg.RetryConfiguration.QueueCapacity,
129130
MinBytes: cfg.Reader.MinBytes,
130131
MaxBytes: cfg.Reader.MaxBytes,
131132
MaxWait: cfg.Reader.MaxWait,
@@ -137,8 +138,10 @@ func (cfg *ConsumerConfig) newCronsumerConfig() *kcronsumer.Config {
137138
RetentionTime: cfg.Reader.RetentionTime,
138139
},
139140
Producer: kcronsumer.ProducerConfig{
140-
Balancer: cfg.RetryConfiguration.Balancer,
141-
Brokers: cfg.RetryConfiguration.Brokers,
141+
Balancer: cfg.RetryConfiguration.Balancer,
142+
Brokers: cfg.RetryConfiguration.Brokers,
143+
BatchSize: cfg.RetryConfiguration.ProducerBatchSize,
144+
BatchTimeout: cfg.RetryConfiguration.ProducerBatchTimeout,
142145
},
143146
LogLevel: lcronsumer.Level(cfg.RetryConfiguration.LogLevel),
144147
}
@@ -225,6 +228,9 @@ type RetryConfiguration struct {
225228
WorkDuration time.Duration
226229
SkipMessageByHeaderFn SkipMessageByHeaderFn
227230
Concurrency int
231+
QueueCapacity int
232+
ProducerBatchSize int
233+
ProducerBatchTimeout time.Duration
228234
}
229235

230236
type BatchConfiguration struct {

‎go.mod

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ module github.com/Trendyol/kafka-konsumer/v2
33
go 1.19
44

55
require (
6-
github.com/Trendyol/kafka-cronsumer v1.5.4
6+
github.com/Trendyol/kafka-cronsumer v1.5.5
77
github.com/Trendyol/otel-kafka-konsumer v0.0.7
88
github.com/ansrivas/fiberprometheus/v2 v2.6.1
99
github.com/gofiber/fiber/v2 v2.52.1

‎go.sum

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
github.com/Trendyol/kafka-cronsumer v1.5.4 h1:r2iyWJ8E+rd5IoRGv/XZ2bKMknxfLh2eApPhMiJodR4=
22
github.com/Trendyol/kafka-cronsumer v1.5.4/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNuJkSxB12CcuZWBNFU=
3+
github.com/Trendyol/kafka-cronsumer v1.5.5-0.20241024185446-b44371b7a3ec h1:82NtbRoJBnH7Qq/9VIMyRbRi6jTc9yytqrpR3KsH5Ks=
4+
github.com/Trendyol/kafka-cronsumer v1.5.5-0.20241024185446-b44371b7a3ec/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNuJkSxB12CcuZWBNFU=
5+
github.com/Trendyol/kafka-cronsumer v1.5.5 h1:R+tZd/A//0NDQbmBwkuMolncOXIeKlqRt1bvnmU6Fn8=
6+
github.com/Trendyol/kafka-cronsumer v1.5.5/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNuJkSxB12CcuZWBNFU=
37
github.com/Trendyol/otel-kafka-konsumer v0.0.7 h1:sT1TE2rgfsdrJWrXKz5j6dPkKJsvP+Tv0Dea4ORqJ+4=
48
github.com/Trendyol/otel-kafka-konsumer v0.0.7/go.mod h1:zdCaFclzRCO9fzcjxkHrWOB3I2+uTPrmkq4zczkD1F0=
59
github.com/andybalholm/brotli v1.0.5 h1:8uQZIdzKmjc/iuPu7O2ioW48L81FgatrcpfFmiq/cCs=

‎test/integration/go.mod

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ require (
1010
)
1111

1212
require (
13-
github.com/Trendyol/kafka-cronsumer v1.5.4 // indirect
13+
github.com/Trendyol/kafka-cronsumer v1.5.5 // indirect
1414
github.com/Trendyol/otel-kafka-konsumer v0.0.7 // indirect
1515
github.com/andybalholm/brotli v1.0.5 // indirect
1616
github.com/ansrivas/fiberprometheus/v2 v2.6.1 // indirect

‎test/integration/go.sum

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@ github.com/Trendyol/kafka-cronsumer v1.5.3/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNu
99
github.com/Trendyol/kafka-cronsumer v1.5.4-0.20240808131305-10cf27589160/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNuJkSxB12CcuZWBNFU=
1010
github.com/Trendyol/kafka-cronsumer v1.5.4-0.20240827135347-7ed5187ea81d/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNuJkSxB12CcuZWBNFU=
1111
github.com/Trendyol/kafka-cronsumer v1.5.4/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNuJkSxB12CcuZWBNFU=
12+
github.com/Trendyol/kafka-cronsumer v1.5.5-0.20241024185446-b44371b7a3ec/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNuJkSxB12CcuZWBNFU=
13+
github.com/Trendyol/kafka-cronsumer v1.5.5/go.mod h1:VpweJmKY+6dppFhzWOZDbZfxBNuJkSxB12CcuZWBNFU=
1214
github.com/Trendyol/otel-kafka-konsumer v0.0.7 h1:sT1TE2rgfsdrJWrXKz5j6dPkKJsvP+Tv0Dea4ORqJ+4=
1315
github.com/Trendyol/otel-kafka-konsumer v0.0.7/go.mod h1:zdCaFclzRCO9fzcjxkHrWOB3I2+uTPrmkq4zczkD1F0=
1416
github.com/andybalholm/brotli v1.0.5 h1:8uQZIdzKmjc/iuPu7O2ioW48L81FgatrcpfFmiq/cCs=

0 commit comments

Comments
 (0)