*: add Writer for relay log

relay/writer/file.go

@@ -216,43 +216,57 @@ func (w *FileWriter) handleFormatDescriptionEvent(ev *replication.BinlogEvent) (
 // handle RotateEvent:
 //   1. update binlog filename if needed
 //   2. write the RotateEvent if not fake
+// NOTE: we only see fake event for RotateEvent in MySQL source code,
+//       if see fake event for other event type, then handle them.
 // NOTE: we do not create a new binlog file when received a RotateEvent,
 //       instead, we create a new binlog file when received a FormatDescriptionEvent.
 //       because a binlog file without any events has no meaning.
-func (w *FileWriter) handleRotateEvent(ev *replication.BinlogEvent) (*Result, error) {
+func (w *FileWriter) handleRotateEvent(ev *replication.BinlogEvent) (result *Result, err error) {
 	rotateEv, ok := ev.Event.(*replication.RotateEvent)
 	if !ok {
 		return nil, errors.NotValidf("except RotateEvent, but got %+v", ev.Header)
 	}
 
-	// update binlog filename if needed
 	var currFile = w.filename.Get()
-	nextFile := string(rotateEv.NextLogName)
-	if nextFile > currFile {
-		// record the next filename, but not create it.
-		// even it's a fake RotateEvent, we still need to record it,
-		// because if we do not specify the filename when creating the writer (like Auto-Position),
-		// we can only receive a fake RotateEvent before the FormatDescriptionEvent.
-		w.filename.Set(nextFile)
-	}
+	defer func() {
+		if err == nil {
+			// update binlog filename if needed
+			nextFile := string(rotateEv.NextLogName)
+			if nextFile > currFile {
+				// record the next filename, but not create it.
+				// even it's a fake RotateEvent, we still need to record it,
+				// because if we do not specify the filename when creating the writer (like Auto-Position),
+				// we can only receive a fake RotateEvent before the FormatDescriptionEvent.
+				w.filename.Set(nextFile)
+			}
+		}
+	}()
 
 	// write the RotateEvent if not fake
-	var ignore bool
 	if ev.Header.Timestamp == 0 || ev.Header.LogPos == 0 {
 		// skip fake rotate event
-		ignore = true
+		return &Result{
+			Ignore: true,
+		}, nil
 	} else if w.out == nil {
 		// if not open a binlog file yet, then non-fake RotateEvent can't be handled
 		return nil, errors.Errorf("non-fake RotateEvent %+v received, but no binlog file opened", ev.Header)
-	} else {
-		err := w.out.Write(ev.RawData)
-		if err != nil {
-			return nil, errors.Annotatef(err, "write RotateEvent %+v for %s", ev.Header, filepath.Join(w.cfg.RelayDir, currFile))
-		}
+	}
+
+	result, err = w.handlePotentialHoleOrDuplicate(ev)
+	if err != nil {
+		return nil, errors.Trace(err)
+	} else if result.Ignore {
+		return result, nil
+	}
+
+	err = w.out.Write(ev.RawData)
+	if err != nil {
+		return nil, errors.Annotatef(err, "write RotateEvent %+v for %s", ev.Header, filepath.Join(w.cfg.RelayDir, currFile))
 	}
 
 	return &Result{
-		Ignore: ignore,
+		Ignore: false,
 	}, nil
 }
 
@@ -261,6 +275,22 @@ func (w *FileWriter) handleRotateEvent(ev *replication.BinlogEvent) (*Result, er
 //   2. handle any duplicate events if exist
 //   3. write the non-duplicate event
 func (w *FileWriter) handleEventDefault(ev *replication.BinlogEvent) (*Result, error) {
+	result, err := w.handlePotentialHoleOrDuplicate(ev)
+	if err != nil {
+		return nil, errors.Trace(err)
+	} else if result.Ignore {
+		return result, nil
+	}
+
+	// write the non-duplicate event
+	err = w.out.Write(ev.RawData)
+	return &Result{
+		Ignore: false,
+	}, errors.Annotatef(err, "write event %+v", ev.Header)
+}
+
+// handlePotentialHoleOrDuplicate combines handleFileHoleExist and handleDuplicateEventsExist.
+func (w *FileWriter) handlePotentialHoleOrDuplicate(ev *replication.BinlogEvent) (*Result, error) {
 	// handle a potential hole
 	mayDuplicate, err := w.handleFileHoleExist(ev)
 	if err != nil {
@@ -281,11 +311,9 @@ func (w *FileWriter) handleEventDefault(ev *replication.BinlogEvent) (*Result, e
 		}
 	}
 
-	// write the non-duplicate event
-	err = w.out.Write(ev.RawData)
 	return &Result{
 		Ignore: false,
-	}, errors.Annotatef(err, "write event %+v", ev.Header)
+	}, nil
 }
 
 // handleFileHoleExist tries to handle a potential hole after this event wrote.
@@ -330,7 +358,7 @@ func (w *FileWriter) handleDuplicateEventsExist(ev *replication.BinlogEvent) (*R
 	filename := filepath.Join(w.cfg.RelayDir, w.filename.Get())
 	duplicate, err := checkIsDuplicateEvent(filename, ev)
 	if err != nil {
-		return nil, errors.Annotatef(err, "check event %+v is duplicate in %s", ev, filename)
+		return nil, errors.Annotatef(err, "check event %+v whether duplicate in %s", ev.Header, filename)
 	}
 
 	return &Result{

relay/writer/file_test.go

@@ -261,6 +261,11 @@ func (t *testFileWriterSuite) TestRotateEventWithFormatDescriptionEvent(c *check
 		latestPos uint32 = 4
 	)
 
+	formatDescEv, err := event.GenFormatDescriptionEvent(header, latestPos)
+	c.Assert(err, check.IsNil)
+	c.Assert(formatDescEv, check.NotNil)
+	latestPos = formatDescEv.Header.LogPos
+
 	rotateEv, err := event.GenRotateEvent(header, latestPos, []byte(nextFilename), nextFilePos)
 	c.Assert(err, check.IsNil)
 	c.Assert(rotateEv, check.NotNil)
@@ -269,9 +274,10 @@ func (t *testFileWriterSuite) TestRotateEventWithFormatDescriptionEvent(c *check
 	c.Assert(err, check.IsNil)
 	c.Assert(fakeRotateEv, check.NotNil)
 
-	formatDescEv, err := event.GenFormatDescriptionEvent(header, latestPos)
+	// hole exists between formatDescEv and holeRotateEv, but the size is too small to fill
+	holeRotateEv, err := event.GenRotateEvent(header, latestPos+event.MinUserVarEventLen-1, []byte(nextFilename), nextFilePos)
 	c.Assert(err, check.IsNil)
-	c.Assert(formatDescEv, check.NotNil)
+	c.Assert(holeRotateEv, check.NotNil)
 
 	// 1: non-fake RotateEvent before FormatDescriptionEvent, invalid
 	w1 := NewFileWriter(cfg, t.parser)
@@ -347,6 +353,11 @@ func (t *testFileWriterSuite) TestRotateEventWithFormatDescriptionEvent(c *check
 	c.Assert(result, check.NotNil)
 	c.Assert(result.Ignore, check.IsFalse)
 
+	// try to write a rotateEv with hole exists
+	result, err = w4.WriteEvent(holeRotateEv)
+	c.Assert(err, check.ErrorMatches, ".*required dummy event size.*is too small.*")
+	c.Assert(result, check.IsNil)
+
 	result, err = w4.WriteEvent(rotateEv)
 	c.Assert(err, check.IsNil)
 	c.Assert(result, check.NotNil)
@@ -355,6 +366,11 @@ func (t *testFileWriterSuite) TestRotateEventWithFormatDescriptionEvent(c *check
 	fileSize += int64(len(rotateEv.RawData))
 	t.verifyFilenameOffset(c, w4, nextFilename, fileSize)
 
+	// write again, duplicate, but we already rotated and new binlog file not created
+	result, err = w4.WriteEvent(rotateEv)
+	c.Assert(err, check.ErrorMatches, ".*no such file or directory.*")
+	c.Assert(result, check.IsNil)
+
 	// cfg.Filename should contain both one FormatDescriptionEvent and one RotateEvent, next file should be empty
 	filename1 = filepath.Join(cfg.RelayDir, cfg.Filename)
 	filename2 = filepath.Join(cfg.RelayDir, nextFilename)