Reset timer when cutting a block via timeout

- Fixes: https://jira.hyperledger.org/browse/FAB-899
- Introduces a BroadcastConsecutiveIncompleteBatches test that failed
before this fix.
- Lowers the BatchTimeout to 500 milliseconds, so as to speed up the
execution of the time-based unit tests in the package (drop from 7 to 3
seconds).
- Moves the timer.Reset() calls before the sendBlock() ones to make the
timing more accurate.

Thanks to Bishop Brock for reporting this.

Change-Id: I9cd8179fa54398f03e0f3b9ef95c080508cce8a3
Signed-off-by: Kostas Christidis <[email protected]>

Author: kchristidis

orderer/kafka/broadcast.go

@@ -98,15 +98,16 @@ func (b *broadcasterImpl) cutBlock(period time.Duration, maxSize uint) {
 		case msg := <-b.batchChan:
 			b.messages = append(b.messages, msg)
 			if len(b.messages) >= int(maxSize) {
-				if err := b.sendBlock(); err != nil {
-					panic(fmt.Errorf("Cannot communicate with Kafka broker: %s", err))
-				}
 				if !timer.Stop() {
 					<-timer.C
 				}
 				timer.Reset(period)
+				if err := b.sendBlock(); err != nil {
+					panic(fmt.Errorf("Cannot communicate with Kafka broker: %s", err))
+				}
 			}
 		case <-timer.C:
+			timer.Reset(period)
 			if len(b.messages) > 0 {
 				if err := b.sendBlock(); err != nil {
 					panic(fmt.Errorf("Cannot communicate with Kafka broker: %s", err))

orderer/kafka/broadcast_test.go

@@ -186,6 +186,59 @@ func TestBroadcastIncompleteBatch(t *testing.T) {
 	}
 }
 
+func TestBroadcastConsecutiveIncompleteBatches(t *testing.T) {
+	if testConf.General.BatchSize <= 1 {
+		t.Skip("Skipping test as it requires a batchsize > 1")
+	}
+
+	messageCount := int(testConf.General.BatchSize) - 1
+
+	disk := make(chan []byte)
+
+	mb := mockNewBroadcaster(t, testConf, oldestOffset, disk)
+	defer testClose(t, mb)
+
+	mbs := newMockBroadcastStream(t)
+	go func() {
+		if err := mb.Broadcast(mbs); err != nil {
+			t.Fatal("Broadcast error:", err)
+		}
+	}()
+
+	for i := 0; i < 2; i++ {
+		<-disk // Checkpoint block in first pass, first incomplete block in second pass -- both tested elsewhere
+
+		// Pump less than batchSize messages into the system
+		go func() {
+			for i := 0; i < messageCount; i++ {
+				mbs.incoming <- &ab.BroadcastMessage{Data: []byte("message " + strconv.Itoa(i))}
+			}
+		}()
+
+		// Ignore the broadcast replies as they have been tested elsewhere
+		for i := 0; i < messageCount; i++ {
+			<-mbs.outgoing
+		}
+	}
+
+	for {
+		select {
+		case in := <-disk:
+			block := new(ab.Block)
+			err := proto.Unmarshal(in, block)
+			if err != nil {
+				t.Fatal("Expected a block on the broker's disk")
+			}
+			if len(block.Messages) != messageCount {
+				t.Fatalf("Expected block to have %d messages instead of %d", messageCount, len(block.Messages))
+			}
+			return
+		case <-time.After(testConf.General.BatchTimeout + timePadding):
+			t.Fatal("Should have received a block by now")
+		}
+	}
+}
+
 func TestBroadcastBatchAndQuitEarly(t *testing.T) {
 	disk := make(chan []byte)
 

orderer/kafka/config_test.go

@@ -38,7 +38,7 @@ var (
 var testConf = &config.TopLevel{
 	General: config.General{
 		OrdererType:   "kafka",
-		BatchTimeout:  2 * time.Second,
+		BatchTimeout:  500 * time.Millisecond,
 		BatchSize:     100,
 		QueueSize:     100,
 		MaxWindowSize: 100,