logSink: revert defaultBufferChanSize, add a trigger to flush logSink…

….units when chennal is full (#2431) (#2446)
pingcap · Aug 3, 2021 · a1976ac · a1976ac
1 parent 2c66164
commit a1976ac
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/cdc/sink/cdclog/file.go b/cdc/sink/cdclog/file.go
@@ -87,7 +87,8 @@ func (ts *tableStream) isEmpty() bool {
 }
 
 func (ts *tableStream) shouldFlush() bool {
-	return ts.sendSize.Load() > maxRowFileSize
+	// if sendSize > 5 MB or data chennal is full, flush it
+	return ts.sendSize.Load() > maxPartFlushSize || ts.sendEvents.Load() == defaultBufferChanSize
 }
 
 func (ts *tableStream) flush(ctx context.Context, sink *logSink) error {

diff --git a/cdc/sink/cdclog/s3.go b/cdc/sink/cdclog/s3.go
@@ -37,7 +37,7 @@ const (
 	maxCompletePartSize = 100 << 20 // rotate row changed event file if one complete file larger than 100Mb
 	maxDDLFlushSize     = 10 << 20  // rotate ddl event file if one complete file larger than 10Mb
 
-	defaultBufferChanSize               = 1280000
+	defaultBufferChanSize               = 20480
 	defaultFlushRowChangedEventDuration = 5 * time.Second // TODO make it as a config
 )
 
@@ -78,7 +78,8 @@ func (tb *tableBuffer) isEmpty() bool {
 }
 
 func (tb *tableBuffer) shouldFlush() bool {
-	return tb.sendSize.Load() > maxPartFlushSize
+	// if sendSize > 5 MB or data chennal is full, flush it
+	return tb.sendSize.Load() > maxPartFlushSize || tb.sendEvents.Load() == defaultBufferChanSize
 }
 
 func (tb *tableBuffer) flush(ctx context.Context, sink *logSink) error {