Activity Tracking for Kafka Publishers

weed/mq/topic/local_partition_offset.go

@@ -28,6 +28,9 @@ func (p *LocalPartition) PublishWithOffset(message *mq_pb.DataMessage, assignOff
 		return 0, fmt.Errorf("failed to add message to buffer: %w", err)
 	}
 
+	// Track publish activity for idle cleanup (consistent with Publish method)
+	p.UpdateActivity()
+
 	// Send to follower if needed (same logic as original Publish)
 	if p.publishFolloweMeStream != nil {
 		if followErr := p.publishFolloweMeStream.Send(&mq_pb.PublishFollowMeRequest{