func TestKafkaZkClient_watchGroupList(t *testing.T) {
	mockZookeeper := helpers.MockZookeeperClient{}

	module := fixtureKafkaZkModule()
	viper.Set("consumer.test.group-whitelist", "test.*")
	module.Configure("test", "consumer.test")
	module.zk = &mockZookeeper

	offsetStat := &zk.Stat{Mtime: 894859}
	newGroupChan := make(chan zk.Event)
	topicExistsChan := make(chan zk.Event)
	newTopicChan := make(chan zk.Event)
	newPartitionChan := make(chan zk.Event)
	newOffsetChan := make(chan zk.Event)
	mockZookeeper.On("ChildrenW", "/consumers").Return([]string{"testgroup"}, offsetStat, func() <-chan zk.Event { return newGroupChan }(), nil)
	mockZookeeper.On("ChildrenW", "/consumers/testgroup/offset").Return([]string{"testtopic"}, offsetStat, func() <-chan zk.Event { return newTopicChan }(), nil)
	mockZookeeper.On("ExistsW", "/consumers/testgroup/offset").Return(true, offsetStat, func() <-chan zk.Event { return topicExistsChan }(), nil)
	mockZookeeper.On("ChildrenW", "/consumers/testgroup/offset/testtopic").Return([]string{"0"}, offsetStat, func() <-chan zk.Event { return newPartitionChan }(), nil)
	mockZookeeper.On("GetW", "/consumers/testgroup/offset/testtopic/0").Return([]byte("81234"), offsetStat, func() <-chan zk.Event { return newOffsetChan }(), nil)
	mockZookeeper.On("GetW", "/consumers/testgroup/owners/testtopic/0").Return([]byte("testowner"), offsetStat, func() <-chan zk.Event { return newOffsetChan }(), nil)

	watchEventChan := make(chan zk.Event)

	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		watchEventChan <- zk.Event{
			Type:  zk.EventNodeChildrenChanged,
			State: zk.StateConnected,
			Path:  "/consumers",
		}

		request := <-module.App.StorageChannel
		assert.Equalf(t, protocol.StorageSetConsumerOffset, request.RequestType, "Expected request sent with type StorageSetConsumerOffset, not %v", request.RequestType)
		assert.Equalf(t, "test", request.Cluster, "Expected request sent with cluster test, not %v", request.Cluster)
		assert.Equalf(t, "testtopic", request.Topic, "Expected request sent with topic testtopic, not %v", request.Topic)
		assert.Equalf(t, int32(0), request.Partition, "Expected request sent with partition 0, not %v", request.Partition)
		assert.Equalf(t, "testgroup", request.Group, "Expected request sent with Group testgroup, not %v", request.Group)
		assert.Equalf(t, int64(81234), request.Offset, "Expected Offset to be 8372, not %v", request.Offset)
		assert.Equalf(t, int64(894859), request.Timestamp, "Expected Timestamp to be 1637, not %v", request.Timestamp)

		newGroupChan <- zk.Event{
			Type:  zk.EventNotWatching,
			State: zk.StateConnected,
			Path:  "/consumers/shouldntgetcalled",
		}
		newTopicChan <- zk.Event{
			Type:  zk.EventNotWatching,
			State: zk.StateConnected,
			Path:  "/consumers/testgroup/offset/shouldntgetcalled",
		}
		topicExistsChan <- zk.Event{
			Type:  zk.EventNotWatching,
			State: zk.StateConnected,
			Path:  "/consumers/testgroup/offset",
		}
		newPartitionChan <- zk.Event{
			Type:  zk.EventNotWatching,
			State: zk.StateConnected,
			Path:  "/consumers/testgroup/offset/testtopic/shouldntgetcalled",
		}
		newOffsetChan <- zk.Event{
			Type:  zk.EventNotWatching,
			State: zk.StateConnected,
			Path:  "/consumers/testgroup/offset/testtopic/1/shouldntgetcalled",
		}

		secondRequest := <-module.App.StorageChannel
		assert.Equalf(t, protocol.StorageSetConsumerOwner, secondRequest.RequestType, "Expected request sent with type StorageSetConsumerOwner, not %v", secondRequest.RequestType)
		assert.Equalf(t, "testowner", secondRequest.Owner, "Expected request sent with Owner testowner, not %v", secondRequest.Owner)
	}()

	module.running.Add(1)
	module.watchGroupList(watchEventChan)
	module.running.Wait()

	mockZookeeper.AssertExpectations(t)
	assert.Equalf(t, int32(1), module.groupList["testgroup"].topics["testtopic"].count, "Expected partition count to be 1, not %v", module.groupList["testgroup"].topics["testtopic"].count)
}
// This tests all the watchers - each one will be called in turn and set, and we assure that they're all closing properly
This tests all the watchers - each one will be called in turn and set, and we assure that they're all closing properly
// This tests all the watchers - each one will be called in turn and set and we assure that they ' re all closing properly
func TestKafkaZkClient_watchGroupList(t *testing.T) {
	mockZookeeper := helpers.MockZookeeperClient{}
	module := fixtureKafkaZkModule()
	viper.Set("consumer.test.group-whitelist", "test.*")
	module.Configure("test", "consumer.test")
	module.zk = &mockZookeeper
	offsetStat := &zk.Stat{Mtime: 894859}
	newGroupChan := make(chan zk.Event)
	topicExistsChan := make(chan zk.Event)
	newTopicChan := make(chan zk.Event)
	newPartitionChan := make(chan zk.Event)
	newOffsetChan := make(chan zk.Event)
	mockZookeeper.On("ChildrenW", "/consumers").Return([]string{"testgroup"}, offsetStat, func() <-chan zk.Event { return newGroupChan }(), nil)
	mockZookeeper.On("ChildrenW", "/consumers/testgroup/offset").Return([]string{"testtopic"}, offsetStat, func() <-chan zk.Event { return newTopicChan }(), nil)
	mockZookeeper.On("ExistsW", "/consumers/testgroup/offset").Return(true, offsetStat, func() <-chan zk.Event { return topicExistsChan }(), nil)
	mockZookeeper.On("ChildrenW", "/consumers/testgroup/offset/testtopic").Return([]string{"0"}, offsetStat, func() <-chan zk.Event { return newPartitionChan }(), nil)
	mockZookeeper.On("GetW", "/consumers/testgroup/offset/testtopic/0").Return([]byte("81234"), offsetStat, func() <-chan zk.Event { return newOffsetChan }(), nil)
	mockZookeeper.On("GetW", "/consumers/testgroup/owners/testtopic/0").Return([]byte("testowner"), offsetStat, func() <-chan zk.Event { return newOffsetChan }(), nil)
	watchEventChan := make(chan zk.Event)
	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		watchEventChan <- zk.Event{
			Type:  zk.EventNodeChildrenChanged,
			State: zk.StateConnected,
			Path:  "/consumers",
		}
		request := <-module.App.StorageChannel
		assert.Equalf(t, protocol.StorageSetConsumerOffset, request.RequestType, "Expected request sent with type StorageSetConsumerOffset, not %v", request.RequestType)
		assert.Equalf(t, "test", request.Cluster, "Expected request sent with cluster test, not %v", request.Cluster)
		assert.Equalf(t, "testtopic", request.Topic, "Expected request sent with topic testtopic, not %v", request.Topic)
		assert.Equalf(t, int32(0), request.Partition, "Expected request sent with partition 0, not %v", request.Partition)
		assert.Equalf(t, "testgroup", request.Group, "Expected request sent with Group testgroup, not %v", request.Group)
		assert.Equalf(t, int64(81234), request.Offset, "Expected Offset to be 8372, not %v", request.Offset)
		assert.Equalf(t, int64(894859), request.Timestamp, "Expected Timestamp to be 1637, not %v", request.Timestamp)
		newGroupChan <- zk.Event{
			Type:  zk.EventNotWatching,
			State: zk.StateConnected,
			Path:  "/consumers/shouldntgetcalled",
		}
		newTopicChan <- zk.Event{
			Type:  zk.EventNotWatching,
			State: zk.StateConnected,
			Path:  "/consumers/testgroup/offset/shouldntgetcalled",
		}
		topicExistsChan <- zk.Event{
			Type:  zk.EventNotWatching,
			State: zk.StateConnected,
			Path:  "/consumers/testgroup/offset",
		}
		newPartitionChan <- zk.Event{
			Type:  zk.EventNotWatching,
			State: zk.StateConnected,
			Path:  "/consumers/testgroup/offset/testtopic/shouldntgetcalled",
		}
		newOffsetChan <- zk.Event{
			Type:  zk.EventNotWatching,
			State: zk.StateConnected,
			Path:  "/consumers/testgroup/offset/testtopic/1/shouldntgetcalled",
		}
		secondRequest := <-module.App.StorageChannel
		assert.Equalf(t, protocol.StorageSetConsumerOwner, secondRequest.RequestType, "Expected request sent with type StorageSetConsumerOwner, not %v", secondRequest.RequestType)
		assert.Equalf(t, "testowner", secondRequest.Owner, "Expected request sent with Owner testowner, not %v", secondRequest.Owner)
	}()
	module.running.Add(1)
	module.watchGroupList(watchEventChan)
	module.running.Wait()
	mockZookeeper.AssertExpectations(t)
	assert.Equalf(t, int32(1), module.groupList["testgroup"].topics["testtopic"].count, "Expected partition count to be 1, not %v", module.groupList["testgroup"].topics["testtopic"].count)
}
func TestKafkaZkClient_watchGroupList ( t * testing . T ) { mockZookeeper := helpers . MockZookeeperClient { } module := fixtureKafkaZkModule ( ) viper . Set ( " " , " " ) module . Configure ( " " , " " ) module . zk = & mockZookeeper offsetStat := & zk . Stat { Mtime : 894859 } newGroupChan := make ( chan zk . Event ) topicExistsChan := make ( chan zk . Event ) newTopicChan := make ( chan zk . Event ) newPartitionChan := make ( chan zk . Event ) newOffsetChan := make ( chan zk . Event ) mockZookeeper . On ( " " , " " ) . Return ( [ ] string { " " } , offsetStat , func ( ) <- chan zk . Event { return newGroupChan } ( ) , nil ) mockZookeeper . On ( " " , " " ) . Return ( [ ] string { " " } , offsetStat , func ( ) <- chan zk . Event { return newTopicChan } ( ) , nil ) mockZookeeper . On ( " " , " " ) . Return ( true , offsetStat , func ( ) <- chan zk . Event { return topicExistsChan } ( ) , nil ) mockZookeeper . On ( " " , " " ) . Return ( [ ] string { " " } , offsetStat , func ( ) <- chan zk . Event { return newPartitionChan } ( ) , nil ) mockZookeeper . On ( " " , " " ) . Return ( [ ] byte ( " " ) , offsetStat , func ( ) <- chan zk . Event { return newOffsetChan } ( ) , nil ) mockZookeeper . On ( " " , " " ) . Return ( [ ] byte ( " " ) , offsetStat , func ( ) <- chan zk . Event { return newOffsetChan } ( ) , nil ) watchEventChan := make ( chan zk . Event ) wg := sync . WaitGroup { } wg . Add ( 1 ) go func ( ) { watchEventChan <- zk . Event { Type : zk . EventNodeChildrenChanged , State : zk . StateConnected , Path : " " , } request := <- module . App . StorageChannel assert . Equalf ( t , protocol . StorageSetConsumerOffset , request . RequestType , " " , request . RequestType ) assert . Equalf ( t , " " , request . Cluster , " " , request . Cluster ) assert . Equalf ( t , " " , request . Topic , " " , request . Topic ) assert . Equalf ( t , int32 ( 0 ) , request . Partition , " " , request . Partition ) assert . Equalf ( t , " " , request . Group , " " , request . Group ) assert . Equalf ( t , int64 ( 81234 ) , request . Offset , " " , request . Offset ) assert . Equalf ( t , int64 ( 894859 ) , request . Timestamp , " " , request . Timestamp ) newGroupChan <- zk . Event { Type : zk . EventNotWatching , State : zk . StateConnected , Path : " " , } newTopicChan <- zk . Event { Type : zk . EventNotWatching , State : zk . StateConnected , Path : " " , } topicExistsChan <- zk . Event { Type : zk . EventNotWatching , State : zk . StateConnected , Path : " " , } newPartitionChan <- zk . Event { Type : zk . EventNotWatching , State : zk . StateConnected , Path : " " , } newOffsetChan <- zk . Event { Type : zk . EventNotWatching , State : zk . StateConnected , Path : " " , } secondRequest := <- module . App . StorageChannel assert . Equalf ( t , protocol . StorageSetConsumerOwner , secondRequest . RequestType , " " , secondRequest . RequestType ) assert . Equalf ( t , " " , secondRequest . Owner , " " , secondRequest . Owner ) } ( ) module . running . Add ( 1 ) module . watchGroupList ( watchEventChan ) module . running . Wait ( ) mockZookeeper . AssertExpectations ( t ) assert . Equalf ( t , int32 ( 1 ) , module . groupList [ " " ] . topics [ " " ] . count , " " , module . groupList [ " " ] . topics [ " " ] . count ) }
This tests all the watchers - each one will be called in turn and set, and we assure that they're all closing properly
This tests all the watchers - each one will be called in turn and set and we assure that they ' re all closing properly