Fix handle duplicate delete message in kafka (uber#1451)

Agent-Tao · Feb 12, 2019 · 99b9eab · 99b9eab
1 parent 8ef3de8
commit 99b9eab
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 0 deletions.
diff --git a/service/worker/indexer/esProcessor.go b/service/worker/indexer/esProcessor.go
@@ -111,6 +111,9 @@ 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)
+	}
 	p.mapToKafkaMsg.Put(key, kafkaMsg)
 	p.processor.Add(request)
 }

diff --git a/service/worker/indexer/esProcessor_test.go b/service/worker/indexer/esProcessor_test.go
@@ -140,6 +140,13 @@ func (s *esProcessorSuite) TestAdd() {
 	s.esProcessor.Add(request, key, mockKafkaMsg)
 	s.Equal(1, s.esProcessor.mapToKafkaMsg.Size())
 	mockKafkaMsg.AssertExpectations(s.T())
+
+	// 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())
 }
 
 func (s *esProcessorSuite) TestBulkAfterAction() {