txnkv: add txn-file docker integration tests and close idle HTTP connections

config/client.go

@@ -121,6 +121,21 @@ type TiKVClient struct {
 
 	// RUV2 is the RU v2 TiKV-side weights used to calculate TiKV RU values from ExecDetailsV2.RuV2.
 	RUV2 RUV2TiKVConfig `toml:"ru-v2" json:"ru-v2"`
+
+	// TxnChunkWriterAddr is the address of the txn chunk writer for file-based txn.
+	TxnChunkWriterAddr string `toml:"txn-chunk-writer-addr" json:"txn-chunk-writer-addr"`
+	// TxnChunkWriterConcurrency is the concurrency to request the txn chunk writer for file-based txn.
+	TxnChunkWriterConcurrency uint `toml:"txn-chunk-writer-concurrency" json:"txn-chunk-writer-concurrency"`
+	// TxnChunkMaxSize is the maximum size of a txn chunk of file-based txn.
+	TxnChunkMaxSize uint64 `toml:"txn-chunk-max-size" json:"txn-chunk-max-size"`
+	// TxnFileMinMutationSize is the minimum size of mutations to use file-based txn.
+	TxnFileMinMutationSize uint64 `toml:"txn-file-min-mutation-size" json:"txn-file-min-mutation-size"`
+	// TxnFileRUDiscountRatio is the discount ratio of resource unit for file-based txn.
+	// Will be ignored if it's <= 0 or >= 1.
+	TxnFileRUDiscountRatio float64 `toml:"txn-file-ru-discount-ratio" json:"txn-file-ru-discount-ratio"`
+	// TxnFileRequestSourceWhitelist is the whitelist of request source types (RequestSource.RequestSourceType) that can use file-based txn.
+	// For internal requests only. External requests can always use file-based txn.
+	TxnFileRequestSourceWhitelist []string `toml:"txn-file-request-source-whitelist" json:"txn-file-request-source-whitelist"`
 }
 
 // RUV2TiKVConfig is the configuration for RU v2 TiKV-side weight calculation.
@@ -232,6 +247,12 @@ func DefaultTiKVClient() TiKVClient {
 		MaxConcurrencyRequestLimit: DefMaxConcurrencyRequestLimit,
 		EnableReplicaSelectorV2:    true,
 		RUV2:                       DefaultRUV2TiKVConfig(),
+
+		TxnChunkWriterConcurrency:     4,
+		TxnChunkMaxSize:               128 * 1024 * 1024,
+		TxnFileMinMutationSize:        16 * 1024 * 1024,
+		TxnFileRUDiscountRatio:        0.125, // filed-based txn costs 1/8 RU of normal txn.
+		TxnFileRequestSourceWhitelist: []string{},
 	}
 }
 
@@ -246,6 +267,22 @@ func (config *TiKVClient) Valid() error {
 	if config.GetGrpcKeepAliveTimeout() < time.Millisecond*50 {
 		return fmt.Errorf("grpc-keepalive-timeout should be at least 0.05, but got %f", config.GrpcKeepAliveTimeout)
 	}
+	return validateTxnFileConfig(config)
+}
+
+func validateTxnFileConfig(config *TiKVClient) error {
+	if config.TxnChunkMaxSize == 0 {
+		return fmt.Errorf("txn-chunk-max-size should be greater than 0")
+	}
+	if config.TxnChunkMaxSize > math.MaxInt {
+		return fmt.Errorf("txn-chunk-max-size should not exceed %d, but got %d", math.MaxInt, config.TxnChunkMaxSize)
+	}
+	if config.TxnChunkWriterConcurrency == 0 {
+		return fmt.Errorf("txn-chunk-writer-concurrency should be greater than 0")
+	}
+	if config.TxnChunkWriterConcurrency > math.MaxInt {
+		return fmt.Errorf("txn-chunk-writer-concurrency should not exceed %d, but got %d", math.MaxInt, config.TxnChunkWriterConcurrency)
+	}
 	return nil
 }
 

config/config_test.go

@@ -35,6 +35,8 @@
 package config
 
 import (
+	"fmt"
+	"math"
 	"testing"
 	"time"
 
@@ -89,3 +91,70 @@ func TestValidateGRPCKeepAliveTimeout(t *testing.T) {
 	assert.NotNil(t, cfg.Valid())
 	assert.Equal(t, "grpc-keepalive-timeout should be at least 0.05, but got 0.040000", cfg.Valid().Error())
 }
+
+func TestValidateTxnFileConfig(t *testing.T) {
+	maxInt := uint64(math.MaxInt)
+	tests := []struct {
+		name      string
+		configure func(*TiKVClient)
+		err       string
+	}{
+		{
+			name: "default",
+		},
+		{
+			name: "zero chunk size",
+			configure: func(cfg *TiKVClient) {
+				cfg.TxnChunkMaxSize = 0
+			},
+			err: "txn-chunk-max-size should be greater than 0",
+		},
+		{
+			name: "maximum chunk size",
+			configure: func(cfg *TiKVClient) {
+				cfg.TxnChunkMaxSize = maxInt
+			},
+		},
+		{
+			name: "chunk size exceeds int",
+			configure: func(cfg *TiKVClient) {
+				cfg.TxnChunkMaxSize = maxInt + 1
+			},
+			err: fmt.Sprintf("txn-chunk-max-size should not exceed %d, but got %d", maxInt, maxInt+1),
+		},
+		{
+			name: "zero writer concurrency",
+			configure: func(cfg *TiKVClient) {
+				cfg.TxnChunkWriterConcurrency = 0
+			},
+			err: "txn-chunk-writer-concurrency should be greater than 0",
+		},
+		{
+			name: "maximum writer concurrency",
+			configure: func(cfg *TiKVClient) {
+				cfg.TxnChunkWriterConcurrency = uint(maxInt)
+			},
+		},
+		{
+			name: "writer concurrency exceeds int",
+			configure: func(cfg *TiKVClient) {
+				cfg.TxnChunkWriterConcurrency = uint(maxInt) + 1
+			},
+			err: fmt.Sprintf("txn-chunk-writer-concurrency should not exceed %d, but got %d", maxInt, uint(maxInt)+1),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			cfg := DefaultTiKVClient()
+			if test.configure != nil {
+				test.configure(&cfg)
+			}
+			if test.err == "" {
+				assert.NoError(t, cfg.Valid())
+				return
+			}
+			assert.EqualError(t, cfg.Valid(), test.err)
+		})
+	}
+}