add PoolCompressor option and pass it to elastictransport.Config (#840)

elastic · Apr 5, 2024 · c5c2b62 · c5c2b62
1 parent 2edc522
commit c5c2b62
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 0 deletions.
diff --git a/elasticsearch.go b/elasticsearch.go
@@ -91,6 +91,7 @@ type Config struct {
 
 	CompressRequestBody      bool // Default: false.
 	CompressRequestBodyLevel int  // Default: gzip.DefaultCompression.
+	PoolCompressor           bool // If true, a sync.Pool based gzip writer is used. Default: false.
 
 	DiscoverNodesOnStart  bool          // Discover nodes when initializing the client. Default: false.
 	DiscoverNodesInterval time.Duration // Discover nodes periodically. Default: disabled.
@@ -294,6 +295,7 @@ func newTransport(cfg Config) (*elastictransport.Client, error) {
 
 		CompressRequestBody:      cfg.CompressRequestBody,
 		CompressRequestBodyLevel: cfg.CompressRequestBodyLevel,
+		PoolCompressor:           cfg.PoolCompressor,
 
 		EnableMetrics:     cfg.EnableMetrics,
 		EnableDebugLogger: cfg.EnableDebugLogger,

diff --git a/esutil/bulk_indexer_integration_test.go b/esutil/bulk_indexer_integration_test.go
@@ -43,6 +43,7 @@ func TestBulkIndexerIntegration(t *testing.T) {
 		name                       string
 		CompressRequestBodyEnabled bool
 		CompressRequestBodyLevel   int
+		PoolCompressor             bool
 	}{
 		{
 			name:                       "Without body compression",
@@ -57,6 +58,11 @@ func TestBulkIndexerIntegration(t *testing.T) {
 			CompressRequestBodyEnabled: true,
 			CompressRequestBodyLevel:   gzip.BestSpeed,
 		},
+		{
+			name:                       "With body compression (sync.Pool)",
+			CompressRequestBodyEnabled: true,
+			PoolCompressor:             true,
+		},
 	}
 
 	for _, tt := range testCases {
@@ -68,6 +74,7 @@ func TestBulkIndexerIntegration(t *testing.T) {
 				es, _ := elasticsearch.NewClient(elasticsearch.Config{
 					CompressRequestBody:      tt.CompressRequestBodyEnabled,
 					CompressRequestBodyLevel: tt.CompressRequestBodyLevel,
+					PoolCompressor:           tt.PoolCompressor,
 					Logger:                   &elastictransport.ColorLogger{Output: os.Stdout},
 				})
 
@@ -134,6 +141,7 @@ func TestBulkIndexerIntegration(t *testing.T) {
 				es, _ := elasticsearch.NewClient(elasticsearch.Config{
 					CompressRequestBody:      tt.CompressRequestBodyEnabled,
 					CompressRequestBodyLevel: tt.CompressRequestBodyLevel,
+					PoolCompressor:           tt.PoolCompressor,
 					Logger:                   &elastictransport.ColorLogger{Output: os.Stdout},
 				})
 
@@ -203,6 +211,7 @@ func TestBulkIndexerIntegration(t *testing.T) {
 				es, _ := elasticsearch.NewClient(elasticsearch.Config{
 					CompressRequestBody:      tt.CompressRequestBodyEnabled,
 					CompressRequestBodyLevel: tt.CompressRequestBodyLevel,
+					PoolCompressor:           tt.PoolCompressor,
 					Logger:                   &elastictransport.ColorLogger{Output: os.Stdout},
 				})
 
@@ -265,6 +274,7 @@ func TestBulkIndexerIntegration(t *testing.T) {
 				es, _ := elasticsearch.NewClient(elasticsearch.Config{
 					CompressRequestBody:      tt.CompressRequestBodyEnabled,
 					CompressRequestBodyLevel: tt.CompressRequestBodyLevel,
+					PoolCompressor:           tt.PoolCompressor,
 					Logger:                   &elastictransport.ColorLogger{Output: os.Stdout, EnableRequestBody: true, EnableResponseBody: true},
 				})