Merge pull request #77 from travisjeffery/tests

more broker tests

Author: Travis Jeffery

broker/broker.go

@@ -256,8 +256,7 @@ func (b *Broker) handleLeaderAndISR(header *protocol.RequestHeader, req *protoco
 	}
 	for i, p := range req.PartitionStates {
 		partition, err := b.partition(p.Topic, p.Partition)
-		// TODO: seems ok to have protocol.ErrUnknownTopicOrPartition here?
-		if err != protocol.ErrNone {
+		if err != protocol.ErrUnknownTopicOrPartition && err != protocol.ErrNone {
 			setErr(i, p, err)
 			continue
 		}
@@ -289,6 +288,7 @@ func (b *Broker) handleLeaderAndISR(header *protocol.RequestHeader, req *protoco
 				continue
 			}
 		}
+		resp.Partitions[i] = &protocol.LeaderAndISRPartition{Partition: p.Partition, Topic: p.Topic, ErrorCode: protocol.ErrNone.Code()}
 	}
 	return resp
 }

broker/broker_test.go

@@ -444,6 +444,40 @@ func TestBroker_Run(t *testing.T) {
 					}}}},
 			},
 		},
+		{
+			name:   "leader and isr leader new partition",
+			fields: newFields(),
+			args: args{
+				requestCh:  make(chan jocko.Request, 2),
+				responseCh: make(chan jocko.Response, 2),
+				requests: []jocko.Request{{
+					Header: &protocol.RequestHeader{CorrelationID: 2},
+					Request: &protocol.LeaderAndISRRequest{
+						PartitionStates: []*protocol.PartitionState{
+							{
+								Topic:     "the-topic",
+								Partition: 1,
+								ISR:       []int32{1},
+								Replicas:  []int32{1},
+								Leader:    1,
+								ZKVersion: 1,
+							},
+						},
+					}},
+				},
+				responses: []jocko.Response{{
+					Header: &protocol.RequestHeader{CorrelationID: 2},
+					Response: &protocol.Response{CorrelationID: 2, Body: &protocol.LeaderAndISRResponse{
+						Partitions: []*protocol.LeaderAndISRPartition{
+							{
+								ErrorCode: protocol.ErrNone.Code(),
+								Partition: 1,
+								Topic:     "the-topic",
+							},
+						},
+					}}}},
+			},
+		},
 	}
 	for _, tt := range tests {
 		os.RemoveAll("/tmp/jocko")
@@ -1225,42 +1259,32 @@ func TestBroker_deletePartitions(t *testing.T) {
 }
 
 func TestBroker_Shutdown(t *testing.T) {
-	type fields struct {
-		logger      *simplelog.Logger
-		id          int32
-		topicMap    map[string][]*jocko.Partition
-		replicators map[*jocko.Partition]*Replicator
-		brokerAddr  string
-		logDir      string
-		raft        jocko.Raft
-		serf        jocko.Serf
-		shutdownCh  chan struct{}
-		shutdown    bool
-	}
 	tests := []struct {
 		name    string
 		fields  fields
 		wantErr bool
 	}{
-	// TODO: Add test cases.
+		{
+			name:    "shutdown ok",
+			fields:  newFields(),
+			wantErr: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			b := &Broker{
-				logger:      tt.fields.logger,
-				id:          tt.fields.id,
-				topicMap:    tt.fields.topicMap,
-				replicators: tt.fields.replicators,
-				brokerAddr:  tt.fields.brokerAddr,
-				logDir:      tt.fields.logDir,
-				raft:        tt.fields.raft,
-				serf:        tt.fields.serf,
-				shutdownCh:  tt.fields.shutdownCh,
-				shutdown:    tt.fields.shutdown,
+			b, err := New(tt.fields.id, Addr(tt.fields.brokerAddr), Serf(tt.fields.serf), Raft(tt.fields.raft), Logger(tt.fields.logger), RaftCommands(tt.fields.raftCommands), LogDir(tt.fields.logDir))
+			if err != nil {
+				t.Errorf("Broker.New() error = %v, wanted nil", err)
 			}
 			if err := b.Shutdown(); (err != nil) != tt.wantErr {
 				t.Errorf("Broker.Shutdown() error = %v, wantErr %v", err, tt.wantErr)
 			}
+			if tt.fields.raft.ShutdownInvoked != true {
+				t.Errorf("did not shutdown raft")
+			}
+			if tt.fields.serf.ShutdownInvoked != true {
+				t.Errorf("did not shutdown raft")
+			}
 		})
 	}
 }
@@ -1415,6 +1439,9 @@ func newFields() fields {
 		MemberFn: func(memberID int32) *jocko.ClusterMember {
 			return &jocko.ClusterMember{ID: 1}
 		},
+		ShutdownFn: func() error {
+			return nil
+		},
 	}
 	raft := &mock.Raft{
 		AddrFn: func() string {
@@ -1438,6 +1465,9 @@ func newFields() fields {
 		ApplyFn: func(jocko.RaftCommand) error {
 			return nil
 		},
+		ShutdownFn: func() error {
+			return nil
+		},
 	}
 	return fields{
 		topicMap:     make(map[string][]*jocko.Partition),