minor change (#295)

Author: nischitpra

configs/config.go

@@ -289,6 +289,7 @@ type Config struct {
 	CommitterKafkaEnableTLS          bool   `env:"COMMITTER_KAFKA_ENABLE_TLS" envDefault:"true"`
 	CommitterMaxMemoryMB             int    `env:"COMMITTER_MAX_MEMORY_MB" envDefault:"512"`
 	CommitterCompressionThresholdMB  int    `env:"COMMITTER_COMPRESSION_THRESHOLD_MB" envDefault:"50"`
+	CommitterKafkaBatchSize          int    `env:"COMMITTER_KAFKA_BATCH_SIZE" envDefault:"500"`
 	StagingS3Bucket                  string `env:"STAGING_S3_BUCKET" envDefault:"thirdweb-insight-production"`
 	StagingS3Region                  string `env:"STAGING_S3_REGION" envDefault:"us-west-2"`
 	StagingS3AccessKeyID             string `env:"STAGING_S3_ACCESS_KEY_ID"`

internal/committer/blockprocessorroutine.go

@@ -19,7 +19,7 @@ func processBlocks() {
 	indexerName := config.Cfg.ZeetProjectName
 
 	totalBytesInBatch := uint64(0)
-	blockBatch := make([]*common.BlockData, 0, 500)
+	blockBatch := make([]*common.BlockData, 0, config.Cfg.CommitterKafkaBatchSize)
 	defer func() {
 		releaseMemoryPermit(totalBytesInBatch)
 	}()
@@ -37,7 +37,7 @@ func processBlocks() {
 		if block.Acquired {
 			totalBytesInBatch += block.ByteSize
 		}
-		if len(blockBatch) == 500 {
+		if len(blockBatch) == config.Cfg.CommitterKafkaBatchSize {
 			if err := libs.KafkaPublisherV2.PublishBlockData(blockBatch); err != nil {
 				log.Panic().
 					Err(err).
@@ -56,7 +56,7 @@ func processBlocks() {
 				Uint64("memory_released_bytes", totalBytesInBatch).
 				Msg("Successfully published batch to Kafka")
 
-			blockBatch = make([]*common.BlockData, 0, 500)
+			blockBatch = make([]*common.BlockData, 0, config.Cfg.CommitterKafkaBatchSize)
 		}
 
 		nextBlockNumber++