Ensure ordering of offset commit requests

Add locking to prevent concurrent calls to commit from potentially being
re-ordered between determining which offsets to include in the offset
commit request, and sending the request to Kafka.

Fixes: IBM#2940
Co-authored-by: Michael Burgess <michburg@uk.ibm.com>
Signed-off-by: Adrian Preston <PRESTONA@uk.ibm.com>

offset_manager.go

@@ -42,6 +42,8 @@ type offsetManager struct {
 	poms     map[string]map[int32]*partitionOffsetManager
 	pomsLock sync.RWMutex
 
+	commitLock sync.Mutex
+
 	closeOnce sync.Once
 	closing   chan none
 	closed    chan none
@@ -251,17 +253,19 @@ func (om *offsetManager) Commit() {
 }
 
 func (om *offsetManager) flushToBroker() {
-	req := om.constructRequest()
-	if req == nil {
-		return
-	}
-
 	broker, err := om.coordinator()
 	if err != nil {
 		om.handleError(err)
 		return
 	}
 
+	om.commitLock.Lock()
+	defer om.commitLock.Unlock()
+	req := om.constructRequest()
+	if req == nil {
+		return
+	}
+
 	resp, err := broker.CommitOffset(req)
 	if err != nil {
 		om.handleError(err)