This is an automated cherry-pick of #63099

Signed-off-by: ti-chi-bot <[email protected]>

Author: tangenta

br/pkg/lightning/backend/local/rate_limiter_param.go

@@ -46,33 +46,29 @@ var (
 // InitializeRateLimiterParam initializes the rate limiter params.
 func InitializeRateLimiterParam(m *meta.Meta, logger *zap.Logger) error {
 	err := initializeVariables(
-		m.GetIngestMaxBatchSplitRanges,
-		defaultMaxBatchSplitRanges,
-		&CurrentMaxBatchSplitRanges,
+		m.GetIngestMaxBatchSplitRanges, m.SetIngestMaxBatchSplitRanges,
+		defaultMaxBatchSplitRanges, &CurrentMaxBatchSplitRanges,
 		logger, "maxBatchSplitRanges")
 	if err != nil {
 		return err
 	}
 	err = initializeVariables(
-		m.GetIngestMaxSplitRangesPerSec,
-		defaultSplitRangesPerSec,
-		&CurrentMaxSplitRangesPerSec,
+		m.GetIngestMaxSplitRangesPerSec, m.SetIngestMaxSplitRangesPerSec,
+		defaultSplitRangesPerSec, &CurrentMaxSplitRangesPerSec,
 		logger, "maxSplitRangesPerSec")
 	if err != nil {
 		return err
 	}
 	err = initializeVariables(
-		m.GetIngestMaxInflight,
-		defaultMaxIngestInflight,
-		&CurrentMaxIngestInflight,
+		m.GetIngestMaxInflight, m.SetIngestMaxInflight,
+		defaultMaxIngestInflight, &CurrentMaxIngestInflight,
 		logger, "maxIngestInflight")
 	if err != nil {
 		return err
 	}
 	err = initializeVariables(
-		m.GetIngestMaxPerSec,
-		defaultMaxIngestPerSec,
-		&CurrentMaxIngestPerSec,
+		m.GetIngestMaxPerSec, m.SetIngestMaxPerSec,
+		defaultMaxIngestPerSec, &CurrentMaxIngestPerSec,
 		logger, "maxIngestPerSec")
 	if err != nil {
 		return err
@@ -82,6 +78,7 @@ func InitializeRateLimiterParam(m *meta.Meta, logger *zap.Logger) error {
 
 func initializeVariables[T comparable](
 	metaGetter func() (v T, isNull bool, err error),
+	metaSetter func(v T) error,
 	defaultVal T,
 	globalVar *atomic.Pointer[T],
 	logger *zap.Logger,
@@ -92,11 +89,17 @@ func initializeVariables[T comparable](
 		return errors.Annotatef(err, "failed to read %s value from meta store", varName)
 	}
 	var zero T
-	if isNull || val == zero {
+	if isNull {
+		err = metaSetter(defaultVal)
+		if err != nil {
+			return errors.Annotatef(err, "failed to set %s value to meta store", varName)
+		}
 		val = defaultVal
 		logger.Info("meta kv not found in meta store, initialized to default and persisted",
 			zap.String("key", varName),
 			zap.Any("value", defaultVal))
+	} else if val == zero {
+		val = defaultVal
 	} else {
 		logger.Info("loaded value from meta store",
 			zap.String("key", varName),

pkg/server/handler/tests/BUILD.bazel

@@ -9,8 +9,12 @@ go_test(
         "main_test.go",
     ],
     flaky = True,
+<<<<<<< HEAD
     race = "on",
     shard_count = 38,
+=======
+    shard_count = 41,
+>>>>>>> 63614029658 (server/handler: register missing http handlers related to ingest param (#63099))
     deps = [
         "//pkg/config",
         "//pkg/ddl",

pkg/server/handler/tests/http_handler_serial_test.go

@@ -684,3 +684,81 @@ func TestTTL(t *testing.T) {
 	require.Nil(t, obj)
 	require.EqualError(t, err, "http status: 400 Bad Request, table test_ttl.t2 not exists")
 }
+<<<<<<< HEAD
+=======
+
+func TestGC(t *testing.T) {
+	ts := createBasicHTTPHandlerTestSuite()
+	ts.startServer(t)
+	defer ts.stopServer(t)
+
+	var data url.Values
+	resp, err := ts.FormStatus("/txn-gc-states", data)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusMethodNotAllowed, resp.StatusCode)
+
+	resp, err = ts.FetchStatus("/txn-gc-states")
+	require.NoError(t, err)
+	defer func() { require.NoError(t, resp.Body.Close()) }()
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+
+	// Verify the resp body.
+	decoder := json.NewDecoder(resp.Body)
+	var state gc.GCState
+	err = decoder.Decode(&state)
+	require.NoError(t, err)
+
+	var empty gc.GCState
+	require.NotEqual(t, empty, state)
+}
+
+func TestIngestParam(t *testing.T) {
+	ts := createBasicHTTPHandlerTestSuite()
+	ts.startServer(t)
+	defer ts.stopServer(t)
+
+	testCases := []struct {
+		url         string
+		defaultVal  any
+		modVal      any
+		expectedVal any
+	}{
+		{"/ingest/max-batch-split-ranges", float64(2048), 1000, float64(1000)},
+		{"/ingest/max-split-ranges-per-sec", float64(0), 2000, float64(2000)},
+		{"/ingest/max-ingest-inflight", float64(0), 1000, float64(1000)},
+		{"/ingest/max-ingest-per-sec", float64(0), 2000, float64(2000)},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.url, func(t *testing.T) {
+			resp, err := ts.FetchStatus(tc.url)
+			require.NoError(t, err)
+			defer func() { require.NoError(t, resp.Body.Close()) }()
+			require.Equal(t, http.StatusOK, resp.StatusCode)
+			decoder := json.NewDecoder(resp.Body)
+			var payload struct {
+				Value  float64 `json:"value"`
+				IsNull bool    `json:"is_null"`
+			}
+			err = decoder.Decode(&payload)
+			require.NoError(t, err)
+			require.Equal(t, tc.defaultVal, payload.Value)
+
+			resp, err = ts.PostStatus(tc.url, "", bytes.NewBuffer([]byte(fmt.Sprintf(`{"value": %v}`, tc.modVal))))
+			require.NoError(t, err)
+			require.NotNil(t, resp)
+			defer func() { require.NoError(t, resp.Body.Close()) }()
+			require.Equal(t, http.StatusOK, resp.StatusCode)
+
+			resp, err = ts.FetchStatus(tc.url)
+			require.NoError(t, err)
+			defer func() { require.NoError(t, resp.Body.Close()) }()
+			require.Equal(t, http.StatusOK, resp.StatusCode)
+			decoder = json.NewDecoder(resp.Body)
+			err = decoder.Decode(&payload)
+			require.NoError(t, err)
+			require.Equal(t, tc.expectedVal, payload.Value)
+		})
+	}
+}
+>>>>>>> 63614029658 (server/handler: register missing http handlers related to ingest param (#63099))

pkg/server/http_status.go

@@ -257,7 +257,21 @@ func (s *Server) startHTTPServer() {
 	// HTTP path for upgrade operations.
 	router.Handle("/upgrade/{op}", handler.NewClusterUpgradeHandler(tikvHandlerTool.Store.(kv.Storage))).Name("upgrade operations")
 
+<<<<<<< HEAD
 	if s.cfg.Store == "tikv" {
+=======
+	// HTTP path for ingest configurations
+	router.Handle("/ingest/max-batch-split-ranges", tikvhandler.NewIngestConcurrencyHandler(
+		tikvHandlerTool, tikvhandler.IngestParamMaxBatchSplitRanges)).Name("IngestMaxBatchSplitRanges")
+	router.Handle("/ingest/max-split-ranges-per-sec", tikvhandler.NewIngestConcurrencyHandler(
+		tikvHandlerTool, tikvhandler.IngestParamMaxSplitRangesPerSec)).Name("IngestMaxSplitRangesPerSec")
+	router.Handle("/ingest/max-ingest-inflight", tikvhandler.NewIngestConcurrencyHandler(
+		tikvHandlerTool, tikvhandler.IngestParamMaxInflight)).Name("IngestMaxInflight")
+	router.Handle("/ingest/max-ingest-per-sec", tikvhandler.NewIngestConcurrencyHandler(
+		tikvHandlerTool, tikvhandler.IngestParamMaxPerSecond)).Name("IngestMaxPerSec")
+
+	if s.cfg.Store == config.StoreTypeTiKV {
+>>>>>>> 63614029658 (server/handler: register missing http handlers related to ingest param (#63099))
 		// HTTP path for tikv.
 		router.Handle("/tables/{db}/{table}/regions", tikvhandler.NewTableHandler(tikvHandlerTool, tikvhandler.OpTableRegions))
 		router.Handle("/tables/{db}/{table}/ranges", tikvhandler.NewTableHandler(tikvHandlerTool, tikvhandler.OpTableRanges))