Merge pull request #18 from r--w/master

Handling errors in bulk indexing.

goes.go

@@ -350,6 +350,17 @@ func (req *Request) Run() (Response, error) {
 		return Response{}, err
 	}
 
+	if req.api == "_bulk" && esResp.Errors {
+		for _, item := range esResp.Items {
+			for _, i := range item {
+				if i.Error != "" {
+					return Response{}, &SearchError{i.Error, i.Status}
+				}
+			}
+		}
+		return Response{}, &SearchError{Msg: "Unknown error while bulk indexing"}
+	}
+
 	if esResp.Error != "" {
 		return Response{}, &SearchError{esResp.Error, esResp.Status}
 	}

goes_test.go

@@ -227,6 +227,7 @@ func (s *GoesTestSuite) TestBulkSend(c *C) {
 		Type:    docType,
 		Version: 1,
 		Index:   indexName,
+		Status:  201, //201 for indexing ( https://issues.apache.org/jira/browse/CONNECTORS-634 )
 	}
 	c.Assert(response.Items[0][BULK_COMMAND_INDEX], Equals, i)
 	c.Assert(err, IsNil)
@@ -283,6 +284,7 @@ func (s *GoesTestSuite) TestBulkSend(c *C) {
 		Type:    docType,
 		Version: 2,
 		Index:   indexName,
+		Status:  200, //200 for updates
 	}
 	c.Assert(response.Items[0][BULK_COMMAND_DELETE], Equals, i)
 

structs.go

@@ -59,6 +59,7 @@ type Request struct {
 type Response struct {
 	Acknowledged bool
 	Error        string
+	Errors       bool
 	Status       uint64
 	Took         uint64
 	TimedOut     bool  `json:"timed_out"`
@@ -111,6 +112,8 @@ type Item struct {
 	Id      string `json:"_id"`
 	Index   string `json:"_index"`
 	Version int    `json:"_version"`
+	Error   string `json:"error"`
+	Status  uint64 `json:"status"`
 }
 
 // Represents the "_all" field when calling the _stats API