# Conflicts:
#	subscriber.go

@@ -166,0 +177,4 @@
+						s.kopts.Logger.Debug(ctx, fmt.Sprintf("[kgo] consumer stopped topic %s partition %d", t, p))
+					}
+				case <-timeout.C:
+					s.kopts.Logger.Error(ctx, fmt.Sprintf("[kgo] timeout waiting for consumer topic %s partition %d", t, p))

@@ -305,3 +328,1 @@
-					err = h(pm)
-				case func([]broker.Message) error:
-					err = h([]broker.Message{pm})
+				processCtx, cancel := context.WithTimeout(ctx, 30*time.Second) //waiting process consumer mb set to opts/cfg TimeoutProccesWaitingHandle

@@ -308,0 +340,4 @@
+				select {
+				case err = <-errChan:
+				case <-processCtx.Done():
+					err = fmt.Errorf("[kgo] message processing timeout topic %s partition %d offset %d", record.Topic, record.Partition, record.Offset)

@@ -335,6 +372,8 @@ func (pc *consumer) consume() {
 					continue
 				}
 
+				pc.kopts.Logger.Error(pc.kopts.Context, fmt.Sprintf("[kgo] message not acknowledged topic %s partition %d offset %d", record.Topic, record.Partition, record.Offset))