Skip to content

Commit

Permalink
Fix race for handling duplicate kafka message (cadence-workflow#1475)
Browse files Browse the repository at this point in the history
* Fix race for handling duplicate kafka message

* Fix unit test
  • Loading branch information
vancexu authored Feb 19, 2019
1 parent 8d6adbf commit 2079852
Show file tree
Hide file tree
Showing 2 changed files with 2 additions and 2 deletions.
3 changes: 2 additions & 1 deletion service/worker/indexer/esProcessor.go
Original file line number Diff line number Diff line change
Expand Up @@ -112,7 +112,8 @@ func (p *esProcessorImpl) Stop() {
// Add an ES request, and an map item for kafka message
func (p *esProcessorImpl) Add(request elastic.BulkableRequest, key string, kafkaMsg messaging.Message) {
if p.mapToKafkaMsg.Contains(key) { // handle duplicate delete message
p.ackKafkaMsg(key)
kafkaMsg.Ack()
return
}
p.mapToKafkaMsg.Put(key, kafkaMsg)
p.processor.Add(request)
Expand Down
1 change: 0 additions & 1 deletion service/worker/indexer/esProcessor_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,6 @@ func (s *esProcessorSuite) TestAdd() {

// handle duplicate
mockKafkaMsg.On("Ack").Return(nil).Once()
s.mockBulkProcessor.On("Add", request).Return().Once()
s.esProcessor.Add(request, key, mockKafkaMsg)
s.Equal(1, s.esProcessor.mapToKafkaMsg.Size())
mockKafkaMsg.AssertExpectations(s.T())
Expand Down

0 comments on commit 2079852

Please sign in to comment.