fix: comsumer segment without producer (#89)

* fix: comsumer segment without producer

* docs: update release notes

* fix: check queue reader and comsumer empty queue add some logs

---------

Co-authored-by: hardy <[email protected]>

Author: luohoufu

docs/content.en/docs/release-notes/_index.md

@@ -17,6 +17,7 @@ Information about release notes of INFINI Framework is provided here.
 - Add utility to securely marshal JSON (#85)
 
 ### Bug fix
+- Fixed comsumer segment without producer (#89)
 
 ### Improvements
 - Structure http error response (#86)

modules/queue/disk_queue/consumer.go

@@ -142,6 +142,10 @@ READ_MSG:
 		}
 	}
 
+	// check reader
+	if d.reader == nil {
+		return messages, false, errors.New("reader is nil")
+	}
 	//read message size
 	err = binary.Read(d.reader, binary.BigEndian, &msgSize)
 	if err != nil {
@@ -344,7 +348,7 @@ READ_MSG:
 			}
 			newData, err := zstd.ZSTDDecompress(nil, readBuf)
 			if err != nil {
-				log.Error(err)
+				log.Errorf("decompress message error: %v %v,%v %v", d.fileName, d.segment, d.readPos, err)
 				ctx.UpdateNextOffset(d.segment, nextReadPos)
 				return messages, false, err
 			}
@@ -467,6 +471,11 @@ func (d *Consumer) ResetOffset(segment, readPos int64) error {
 		log.Debugf("reset offset: %v,%v, file: %v, queue:%v", segment, readPos, d.fileName, d.queue)
 	}
 
+	// no producer write data to this queue
+	if d.diskQueue.writeSegmentNum == 0 && d.diskQueue.writePos == 0 {
+		return nil
+	}
+
 	if segment > d.diskQueue.writeSegmentNum {
 		log.Errorf("reading segment [%v] is greater than writing segment [%v]", segment, d.diskQueue.writeSegmentNum)
 		return io.EOF

modules/queue/disk_queue/diskqueue.go

@@ -504,6 +504,7 @@ func (d *DiskBasedQueue) readOne() ([]byte, error) {
 		}
 		newData, err := zstd.ZSTDDecompress(nil, readBuf)
 		if err != nil {
+			log.Errorf("diskqueue(%s) failed to decompress %v,%v - %s", d.name, d.readSegmentFileNum, d.readPos)
 			return nil, err
 		}
 		return newData, nil
@@ -552,6 +553,7 @@ func (d *DiskBasedQueue) writeOne(data []byte) WriteResponse {
 		}
 		newData, err := zstd.ZSTDCompress(nil, data, d.cfg.Compress.Message.Level)
 		if err != nil {
+			log.Errorf("diskqueue(%s) failed to compress %v,%v - %s", d.name, d.readSegmentFileNum, d.readPos)
 			res.Error = err
 			return res
 		}

modules/queue/disk_queue/module.go

@@ -374,6 +374,10 @@ func (module *DiskQueue) AcquireConsumer(qconfig *queue.QueueConfig, consumer *q
 	}
 	if ok {
 		q1 := q.(*DiskBasedQueue)
+		if q1.writeSegmentNum == 0 && q1.writePos == 0 {
+			//empty queue, no need to create consumer
+			return nil, errors.New("empty queue")
+		}
 		return q1.AcquireConsumer(qconfig, consumer, offset)
 	}
 	panic(errors.Errorf("queue [%v] not found", qconfig.Name))

plugins/queue/consumer/consumer.go

@@ -485,8 +485,10 @@ func (processor *QueueConsumerProcessor) NewSlicedWorker(ctx *pipeline.Context,
 				case string:
 					v = r.(string)
 				}
-				log.Errorf("worker[%v], queue:[%v], slice:[%v], offset:[%v]->[%v],%v", workerID, qConfig.ID, sliceID, initOffset, offset, v)
-				ctx.Failed(fmt.Errorf("panic in slice worker: %+v", r))
+				if v != "empty queue" {
+					log.Errorf("worker[%v], queue:[%v], slice:[%v], offset:[%v]->[%v],%v", workerID, qConfig.ID, sliceID, initOffset, offset, v)
+					ctx.Failed(fmt.Errorf("panic in slice worker: %+v", r))
+				}
 				if parentContext != nil {
 					parentContext.RecordError(fmt.Errorf("panic in slice worker: %+v", r))
 				}