send flush message to follower before shutting down logBuffer

weed/mq/broker/broker_grpc_pub_follow.go

@@ -43,6 +43,10 @@ func (b *MessageQueueBroker) PublishFollowMe(stream mq_pb.SeaweedMessaging_Publi
 		} else if closeMessage := req.GetClose(); closeMessage != nil {
 			glog.V(0).Infof("topic %v partition %v publish stream closed: %v", initMessage.Topic, initMessage.Partition, closeMessage)
 			break
+		} else if flushMessage := req.GetFlush(); flushMessage != nil {
+			glog.V(0).Infof("topic %v partition %v publish stream flushed: %v", initMessage.Topic, initMessage.Partition, flushMessage)
+		} else {
+			glog.Errorf("unknown message: %v", req)
 		}
 	}
 	return nil

weed/mq/broker/broker_topic_partition_read_write.go

@@ -41,6 +41,16 @@ func (b *MessageQueueBroker) genLogFlushFunc(t topic.Topic, partition *mq_pb.Par
 		}
 
 		atomic.StoreInt64(&logBuffer.LastFlushTsNs, stopTime.UnixNano())
+
+		b.accessLock.Lock()
+		defer b.accessLock.Unlock()
+		p := topic.FromPbPartition(partition)
+		if localPartition:=b.localTopicManager.GetLocalPartition(t, p); localPartition!=nil {
+			localPartition.NotifyLogFlushed(logBuffer.LastFlushTsNs)
+		}
+
+		println("flushing at", logBuffer.LastFlushTsNs, "to", targetFile, "size", len(buf))
+
 	}
 }
 

weed/mq/topic/local_partition.go

@@ -190,6 +190,10 @@ func (p *LocalPartition) MaybeConnectToFollowers(initMessage *mq_pb.PublishMessa
 func (p *LocalPartition) MaybeShutdownLocalPartition() (hasShutdown bool) {
 
 	if p.Publishers.Size() == 0 && p.Subscribers.Size() == 0 {
+		p.LogBuffer.ShutdownLogBuffer()
+		for !p.LogBuffer.IsAllFlushed() {
+			time.Sleep(113 * time.Millisecond)
+		}
 		if p.followerStream != nil {
 			// send close to the follower
 			if followErr := p.followerStream.Send(&mq_pb.PublishFollowMeRequest{
@@ -205,7 +209,6 @@ func (p *LocalPartition) MaybeShutdownLocalPartition() (hasShutdown bool) {
 			p.follower = ""
 		}
 
-		p.LogBuffer.ShutdownLogBuffer()
 		hasShutdown = true
 	}
 
@@ -219,3 +222,18 @@ func (p *LocalPartition) Shutdown() {
 	p.LogBuffer.ShutdownLogBuffer()
 	glog.V(0).Infof("local partition %v shutting down", p.Partition)
 }
+
+func (p *LocalPartition) NotifyLogFlushed(flushTsNs int64) {
+	if p.followerStream != nil {
+		if followErr := p.followerStream.Send(&mq_pb.PublishFollowMeRequest{
+			Message: &mq_pb.PublishFollowMeRequest_Flush{
+				Flush: &mq_pb.PublishFollowMeRequest_FlushMessage{
+					TsNs: flushTsNs,
+				},
+			},
+		}); followErr != nil {
+			glog.Errorf("send follower %s flush message: %v", p.follower, followErr)
+		}
+		println("notifying", p.follower, "flushed at", flushTsNs)
+	}
+}