+ kafkaMessages := `1`
+ wg := sync.WaitGroup{}
+ distributeClientMock := NewTestClient(func(req *http.Request) *http.Response {
+ if req.URL.String() == "http://consumerHost/kafkatarget" {
+ assertions.Equal(req.Method, "POST")
+ assertions.Equal(kafkaMessages, getBodyAsString(req, t))
+ assertions.Equal("text/plain", req.Header.Get("Content-Type"))
+ wg.Done()
+ return &http.Response{
+ StatusCode: 200,
+ Body: ioutil.NopCloser(bytes.NewBufferString(`OK`)),
+ Header: make(http.Header), // Must be set to non-nil value or it panics
+ }
+ }
+ t.Error("Wrong call to client: ", req)
+ t.Fail()
+ return nil
+ })
+
+ kafkaTypeDef := config.TypeDefinition{
+ Identity: "type2",
+ KafkaInputTopic: "topic",
+ }
+ kafkaFactoryMock := mocks.KafkaFactory{}
+ kafkaConsumerMock := mocks.KafkaConsumer{}
+ kafkaConsumerMock.On("Commit").Return([]kafka.TopicPartition{}, error(nil))
+ kafkaConsumerMock.On("Subscribe", mock.Anything).Return(error(nil))
+ kafkaConsumerMock.On("ReadMessage", mock.Anything).Return(&kafka.Message{
+ Value: []byte(kafkaMessages),
+ }, error(nil)).Once()
+ kafkaConsumerMock.On("ReadMessage", mock.Anything).Return(nil, fmt.Errorf("Just to stop"))
+ kafkaFactoryMock.On("NewKafkaConsumer", mock.Anything).Return(kafkaConsumerMock, nil)
+ kafkaJobsHandler := newJobsHandler(kafkaTypeDef, "", kafkaFactoryMock, nil, distributeClientMock)
+
+ jobsManager := NewJobsManagerImpl(nil, "", kafkaFactoryMock, distributeClientMock)
+ jobsManager.allTypes["type2"] = TypeData{
+ Identity: "type2",
+ jobsHandler: kafkaJobsHandler,
+ }
+
+ jobsManager.StartJobsForAllTypes()
+
+ kafkaJobInfo := JobInfo{
+ InfoTypeIdentity: "type2",
+ InfoJobIdentity: "job2",
+ TargetUri: "http://consumerHost/kafkatarget",
+ }
+
+ wg.Add(1) // Wait till the distribution has happened
+ err := jobsManager.AddJobFromRESTCall(kafkaJobInfo)
+ assertions.Nil(err)
+
+ if waitTimeout(&wg, 2*time.Second) {