mayExtendBatchTxn checks if we should extend the current batch transaction
(state *replicationState)
| 958 | |
| 959 | // mayExtendBatchTxn checks if we should extend the current batch transaction |
| 960 | func (r *LogicalReplicator) mayExtendBatchTxn(state *replicationState) (bool, delta.FlushReason) { |
| 961 | extend, reason := false, delta.UnknownFlushReason |
| 962 | if state.ongoingBatchTxn { |
| 963 | extend = true |
| 964 | switch { |
| 965 | case state.commitCount == 1: |
| 966 | // This is the first commit message we've received, we commit it immediately to avoid the keepalive flood |
| 967 | extend, reason = false, delta.InitFlushReason |
| 968 | case time.Since(state.lastCommitTime) >= 200*time.Millisecond: |
| 969 | extend, reason = false, delta.TimeTickFlushReason |
| 970 | case state.deltaBufSize >= (128 << 20): // 128MB |
| 971 | extend, reason = false, delta.MemoryLimitFlushReason |
| 972 | } |
| 973 | } |
| 974 | return extend, reason |
| 975 | } |
| 976 | |
| 977 | func (r *LogicalReplicator) commitOngoingTxnIfClean(state *replicationState, reason delta.FlushReason) error { |
| 978 | if state.dirtyTxn && !state.dirtyStream { |