Fix the error that fail to seek the msg stream for kafka

Signed-off-by: SimFG <bang.fu@zilliz.com>
zilliztech · Nov 23, 2023 · 79f3ff5 · 79f3ff5
1 parent 9ca2777
commit 79f3ff5
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/core/reader/replicate_channel_manager.go b/core/reader/replicate_channel_manager.go
@@ -643,7 +643,7 @@ func (r *replicateChannelHandler) handlePack(pack *msgstream.MsgPack) *msgstream
 			}
 			if err != nil {
 				r.sendErrEvent(err)
-				log.Warn("fail to get partition info", zap.Any("msg", msg), zap.Error(err))
+				log.Warn("fail to get partition info", zap.Any("msg", msg.Type()), zap.Error(err))
 				return nil
 			}
 			if pChannel != info.PChannel {
@@ -721,7 +721,7 @@ func newReplicateChannelHandler(ctx context.Context,
 		return nil, err
 	}
 	subPositionType := mqwrapper.SubscriptionPositionUnknown
-	if sourceInfo.SeekPosition != nil {
+	if sourceInfo.SeekPosition == nil {
 		subPositionType = mqwrapper.SubscriptionPositionLatest
 	}
 	err = stream.AsConsumer(ctx, []string{sourceInfo.PChannelName}, sourceInfo.PChannelName+strconv.Itoa(rand.Int()), subPositionType)