remove useless fields

Author: 3AceShowHand

downstreamadapter/dispatcher/basic_dispatcher.go

@@ -165,9 +165,6 @@ type BasicDispatcher struct {
 	// Shared info containing all common configuration and resources
 	sharedInfo *SharedInfo
 
-	eventCollectorBatchCount int
-	eventCollectorBatchBytes int
-
 	// sink is the sink for this dispatcher
 	sink sink.Sink
 
@@ -235,35 +232,31 @@ func NewBasicDispatcher(
 	schemaIDToDispatchers *SchemaIDToDispatchers,
 	skipSyncpointAtStartTs bool,
 	skipDMLAsStartTs bool,
-	eventCollectorBatchCount int,
-	eventCollectorBatchBytes int,
 	currentPDTs uint64,
 	mode int64,
 	sink sink.Sink,
 	sharedInfo *SharedInfo,
 ) *BasicDispatcher {
 	dispatcher := &BasicDispatcher{
-		id:                       id,
-		tableSpan:                tableSpan,
-		isCompleteTable:          common.IsCompleteSpan(tableSpan),
-		startTs:                  startTs,
-		skipSyncpointAtStartTs:   skipSyncpointAtStartTs,
-		skipDMLAsStartTs:         skipDMLAsStartTs,
-		sharedInfo:               sharedInfo,
-		eventCollectorBatchCount: eventCollectorBatchCount,
-		eventCollectorBatchBytes: eventCollectorBatchBytes,
-		sink:                     sink,
-		componentStatus:          newComponentStateWithMutex(heartbeatpb.ComponentState_Initializing),
-		isRemoving:               atomic.Bool{},
-		duringHandleEvents:       atomic.Bool{},
-		blockEventStatus:         BlockEventStatus{blockPendingEvent: nil},
-		tableProgress:            NewTableProgress(),
-		schemaID:                 schemaID,
-		schemaIDToDispatchers:    schemaIDToDispatchers,
-		resendTaskMap:            newResendTaskMap(),
-		creationPDTs:             currentPDTs,
-		mode:                     mode,
-		BootstrapState:           BootstrapFinished,
+		id:                     id,
+		tableSpan:              tableSpan,
+		isCompleteTable:        common.IsCompleteSpan(tableSpan),
+		startTs:                startTs,
+		skipSyncpointAtStartTs: skipSyncpointAtStartTs,
+		skipDMLAsStartTs:       skipDMLAsStartTs,
+		sharedInfo:             sharedInfo,
+		sink:                   sink,
+		componentStatus:        newComponentStateWithMutex(heartbeatpb.ComponentState_Initializing),
+		isRemoving:             atomic.Bool{},
+		duringHandleEvents:     atomic.Bool{},
+		blockEventStatus:       BlockEventStatus{blockPendingEvent: nil},
+		tableProgress:          NewTableProgress(),
+		schemaID:               schemaID,
+		schemaIDToDispatchers:  schemaIDToDispatchers,
+		resendTaskMap:          newResendTaskMap(),
+		creationPDTs:           currentPDTs,
+		mode:                   mode,
+		BootstrapState:         BootstrapFinished,
 	}
 	dispatcher.resolvedTs.Store(startTs)
 

downstreamadapter/dispatcher/basic_dispatcher_active_active_test.go

@@ -158,8 +158,6 @@ func newTestBasicDispatcher(t *testing.T, sinkType common.SinkType, enableActive
 		NewSchemaIDToDispatchers(),
 		false,
 		false,
-		4096,
-		0,
 		200,
 		common.DefaultMode,
 		dispatcherSink.Sink(),

downstreamadapter/dispatcher/basic_dispatcher_info.go

@@ -140,7 +140,7 @@ func (d *BasicDispatcher) GetChangefeedID() common.ChangeFeedID {
 }
 
 func (d *BasicDispatcher) GetEventCollectorBatchConfig() (batchCount int, batchBytes int) {
-	return d.eventCollectorBatchCount, d.eventCollectorBatchBytes
+	return d.sharedInfo.eventCollectorBatchCount, d.sharedInfo.eventCollectorBatchBytes
 }
 
 func (d *BasicDispatcher) GetComponentStatus() heartbeatpb.ComponentState {

downstreamadapter/dispatcher/event_dispatcher.go

@@ -73,8 +73,6 @@ func NewEventDispatcher(
 		schemaIDToDispatchers,
 		skipSyncpointAtStartTs,
 		skipDMLAsStartTs,
-		sharedInfo.eventCollectorBatchCount,
-		sharedInfo.eventCollectorBatchBytes,
 		currentPdTs,
 		common.DefaultMode,
 		sink,

downstreamadapter/dispatcher/redo_dispatcher.go

@@ -56,8 +56,6 @@ func NewRedoDispatcher(
 		schemaIDToDispatchers,
 		skipSyncpointAtStartTs,
 		skipDMLAsStartTs,
-		sharedInfo.eventCollectorBatchCount,
-		sharedInfo.eventCollectorBatchBytes,
 		0,
 		common.RedoMode,
 		sink,

utils/dynstream/event_queue_batcher_test.go

@@ -32,6 +32,13 @@ func TestBatcherSetLimit(t *testing.T) {
 	require.Equal(t, 64, b.config.hardBytes)
 }
 
+func TestNewAreaSettingsWithMaxPendingSizeAndBatchConfigNormalizesBatchConfig(t *testing.T) {
+	settings := NewAreaSettingsWithMaxPendingSizeAndBatchConfig(
+		64*1024*1024, 0, "test", 0, -1,
+	)
+	require.Equal(t, newBatchConfig(0, -1), settings.batchConfig)
+}
+
 func TestBatchByBytes(t *testing.T) {
 	handler := mockHandler{}
 	registry := newAreaBatchConfigRegistry[int](newDefaultBatchConfig())