Merge pull request #2100 from tomyouyou/master

force to flush segments when recovering a message with last non-CleanShutdown.

(cherry picked from commit dd46dd3)

Author: michaelklishin

src/rabbit_queue_index.erl

@@ -626,7 +626,8 @@ init_dirty(CleanShutdown, ContainsCheckFun, State) ->
                   {{Segment = #segment { unacked = UnackedCount }, Dirty},
                    UnackedBytes} =
                       recover_segment(ContainsCheckFun, CleanShutdown,
-                                      segment_find_or_new(Seg, Dir, Segments2)),
+                                      segment_find_or_new(Seg, Dir, Segments2),
+                                      State1#qistate.max_journal_entries),
                   {segment_store(Segment, Segments2),
                    CountAcc + UnackedCount,
                    BytesAcc + UnackedBytes, DirtyCount + Dirty}
@@ -650,7 +651,7 @@ terminate(State = #qistate { journal_handle = JournalHdl,
                                      segments = undefined }}.
 
 recover_segment(ContainsCheckFun, CleanShutdown,
-                Segment = #segment { journal_entries = JEntries }) ->
+                Segment = #segment { journal_entries = JEntries }, MaxJournal) ->
     {SegEntries, UnackedCount} = load_segment(false, Segment),
     {SegEntries1, UnackedCountDelta} =
         segment_plus_journal(SegEntries, JEntries),
@@ -659,7 +660,7 @@ recover_segment(ContainsCheckFun, CleanShutdown,
            {SegmentAndDirtyCount, Bytes}) ->
               {MsgOrId, MsgProps} = parse_pub_record_body(Bin, MsgBin),
               {recover_message(ContainsCheckFun(MsgOrId), CleanShutdown,
-                               Del, RelSeq, SegmentAndDirtyCount),
+                               Del, RelSeq, SegmentAndDirtyCount, MaxJournal),
                Bytes + case IsPersistent of
                            true  -> MsgProps#message_properties.size;
                            false -> 0
@@ -668,15 +669,16 @@ recover_segment(ContainsCheckFun, CleanShutdown,
       {{Segment #segment { unacked = UnackedCount + UnackedCountDelta }, 0}, 0},
       SegEntries1).
 
-recover_message( true,  true,   _Del, _RelSeq, SegmentAndDirtyCount) ->
+recover_message( true,  true,   _Del, _RelSeq, SegmentAndDirtyCount, _MaxJournal) ->
     SegmentAndDirtyCount;
-recover_message( true, false,    del, _RelSeq, SegmentAndDirtyCount) ->
+recover_message( true, false,    del, _RelSeq, SegmentAndDirtyCount, _MaxJournal) ->
     SegmentAndDirtyCount;
-recover_message( true, false, no_del,  RelSeq, {Segment, DirtyCount}) ->
-    {add_to_journal(RelSeq, del, Segment), DirtyCount + 1};
-recover_message(false,     _,    del,  RelSeq, {Segment, DirtyCount}) ->
+recover_message( true, false, no_del,  RelSeq, {Segment, _DirtyCount}, MaxJournal) ->
+    %% force to flush the segment
+    {add_to_journal(RelSeq, del, Segment), MaxJournal + 1}; 
+recover_message(false,     _,    del,  RelSeq, {Segment, DirtyCount}, _MaxJournal) ->
     {add_to_journal(RelSeq, ack, Segment), DirtyCount + 1};
-recover_message(false,     _, no_del,  RelSeq, {Segment, DirtyCount}) ->
+recover_message(false,     _, no_del,  RelSeq, {Segment, DirtyCount}, _MaxJournal) ->
     {add_to_journal(RelSeq, ack,
                     add_to_journal(RelSeq, del, Segment)),
      DirtyCount + 2}.