Summary of Changes

Hello @asddongmen, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request enhances the dynamic stream (dynstream) and event collector components to handle memory pressure more gracefully. It introduces a mechanism to signal dropped events using a new DropEvent type and refactors the memory control logic for better modularity. When the event collector drops certain event types due to memory limits, it now replaces them with a DropEvent and stops processing further events for that specific path.

Highlights

• Introduce DropEvent: A new event type, DropEvent, is introduced (pkg/common/event/drop_event.go) to explicitly signal when an event (like DML, DDL, or Handshake) is dropped due to memory pressure in the event collector. This allows downstream components to be aware of the dropped event.
• Modify OnDrop Interface: The OnDrop method in the dynstream.Handler interface is updated to return interface{} (utils/dynstream/interfaces.go). This enables handlers to return a replacement event (like the new DropEvent) when an event is dropped, rather than just performing an action.
• Implement Event Dropping in EventCollector: The OnDrop handler for events in the eventcollector (downstreamadapter/eventcollector/helper.go) is updated. If a DML, Handshake, or DDL event is dropped due to memory pressure, it now logs the drop and returns a new DropEvent containing information about the dropped event (DispatcherID, Seq, CommitTs). This DropEvent replaces the original event in the stream.
• Refactor Memory Control Algorithms: The memory control logic within dynstream is refactored into an interface (MemoryControlAlgorithm) and specific implementations (PullerMemoryControl, EventCollectorMemoryControl) (utils/dynstream/memory_control_algorithm.go). The AreaSettings now uses an integer type to specify the algorithm, and the areaMemStat holds an instance of the interface (utils/dynstream/interfaces.go, utils/dynstream/memory_control.go).
• Mark Path as Dead on Drop: When an event is dropped and replaced by a DropEvent in the event collector (using the EventCollectorMemoryControl algorithm), the corresponding path in the dynamic stream is marked as 'dead' (utils/dynstream/stream.go, utils/dynstream/memory_control.go). Subsequent events for a dead path are ignored (utils/dynstream/parallel_dynamic_stream.go).

Code Review

This pull request introduces a mechanism to handle event dropping due to memory pressure, primarily within the dynstream and eventcollector components. Key changes include:

• Refactoring the memory control logic in dynstream by introducing a MemoryControlAlgorithm interface and specific implementations (PullerMemoryControl, EventCollectorMemoryControl). This improves modularity.
• Modifying the Handler.OnDrop method signature to return an interface{}, allowing handlers to provide a replacement event (e.g., a DropEvent) when an event is dropped.
• Introducing a new pkg/common/event.DropEvent type to represent events that have been dropped.
• Updating the eventcollector to utilize this new OnDrop mechanism. When memory limits are exceeded for the EventCollectorMemoryControl algorithm, DML, DDL, or Handshake events are replaced with DropEvents, and the corresponding path is marked as dead.
• Adding String() methods to DDL and DML event types for better debuggability.

Overall, the changes seem well-structured to address the problem of event dropping. However, there are a couple of significant concerns regarding a new log.Panic and the deserialization logic for DropEvent that need to be addressed. Several minor points are also noted.