#3 Limit batch by batch byte size

Makefile

@@ -0,0 +1,19 @@
+.PHONY: default
+
+default: run
+
+init:
+	go mod download
+	go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.49.0
+	go install golang.org/x/tools/go/analysis/passes/fieldalignment/cmd/fieldalignment@latest
+
+clean:
+	rm -rf ./build
+	rm -rf mocks
+
+linter:
+	fieldalignment -fix ./...
+	golangci-lint run -c .golangci.yml --timeout=5m -v --fix
+
+run:
+	go run main.go

README.md

@@ -124,8 +124,9 @@ Check out on [go-dcp-client](https://github.com/Trendyol/go-dcp-client#configura
 | `elasticsearch.collectionIndexMapping` | map[string]string | yes      |         |
 | `elasticsearch.typeName`               | string            | yes      |         |
 | `elasticsearch.urls`                   | []string          | yes      |         |
-| `elasticsearch.bulkSize`               | int               | yes      |         |
-| `elasticsearch.bulkTickerDuration`     | time.Duration     | yes      |         |
+| `elasticsearch.batchSizeLimit`         | int               | yes      |         |
+| `elasticsearch.batchTickerDuration`    | time.Duration     | yes      |         |
+| `elasticsearch.batchByteSizeLimit`     | int               | yes      |         |
 
 ---
 

config/config.go

@@ -14,8 +14,8 @@ type Elasticsearch struct {
 	CollectionIndexMapping map[string]string `yaml:"collectionIndexMapping"`
 	TypeName               string            `yaml:"typeName"`
 	Urls                   []string          `yaml:"urls"`
-	BulkSize               int               `yaml:"bulkSize"`
-	BulkTickerDuration     time.Duration     `yaml:"bulkTickerDuration"`
+	BatchSizeLimit         int               `yaml:"batchSizeLimit"`
+	BatchTickerDuration    time.Duration     `yaml:"batchTickerDuration"`
 }
 
 type Config struct {

elasticsearch/bulk/bulk.go

@@ -29,9 +29,10 @@ type Bulk struct {
 	typeName               []byte
 	batch                  []byte
 	batchSize              int
-	batchLimit             int
+	batchSizeLimit         int
 	batchTickerDuration    time.Duration
 	flushLock              sync.Mutex
+	batchByteSizeLimit     int
 }
 
 type Metric struct {
@@ -51,10 +52,10 @@ func NewBulk(
 	}
 
 	bulk := &Bulk{
-		batchTickerDuration: esConfig.BulkTickerDuration,
-		batchTicker:         time.NewTicker(esConfig.BulkTickerDuration),
-		actionCh:            make(chan document.ESActionDocument, esConfig.BulkSize),
-		batchLimit:          esConfig.BulkSize,
+		batchTickerDuration: esConfig.BatchTickerDuration,
+		batchTicker:         time.NewTicker(esConfig.BatchTickerDuration),
+		actionCh:            make(chan document.ESActionDocument, esConfig.BatchSizeLimit),
+		batchSizeLimit:      esConfig.BatchSizeLimit,
 		isClosed:            make(chan bool, 1),
 		logger:              logger,
 		errorLogger:         errorLogger,
@@ -107,7 +108,7 @@ func (b *Bulk) AddAction(
 
 	b.metric.ESConnectorLatency.Add(float64(time.Since(eventTime).Milliseconds()))
 
-	if b.batchSize == b.batchLimit {
+	if b.batchSize == b.batchSizeLimit || len(b.batch) >= b.batchByteSizeLimit {
 		err := b.flushMessages()
 		if err != nil {
 			b.errorLogger.Printf("Bulk writer error %v", err)

example/config.yml

@@ -21,7 +21,8 @@ elasticsearch:
     _default: indexname
   urls:
     - "http://localhost:9200"
-  bulkSize: 1000
-  bulkTickerDuration: 15s
+  batchSizeLimit: 1000
+  batchTickerDuration: 15s
+  batchByteSizeLimit: 1000
 checkpoint:
   type: manual