func (t *Topic) GetMagicCode() int64 {
	return t.magicCode
}
should be protected by the topic lock for all partitions
package nsqd
import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"os"
	"path"
	"strconv"
	"sync"
	"sync/atomic"
	"time"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/youzan/nsq/internal/ext"
	"github.com/youzan/nsq/internal/levellogger"
	"github.com/youzan/nsq/internal/protocol"
	"github.com/youzan/nsq/internal/quantile"
	"github.com/youzan/nsq/internal/util"
	"github.com/youzan/nsq/nsqd/engine"
)
const (
	MAX_TOPIC_PARTITION    = 1023
	HISTORY_STAT_FILE_NAME = ".stat.history.dat"
	slowCost               = time.Millisecond * 50
)
var (
	ErrInvalidMessageID           = errors.New("message id is invalid")
	ErrWriteOffsetMismatch        = errors.New("write offset mismatch")
	ErrOperationInvalidState      = errors.New("the operation is not allowed under current state")
	ErrMessageInvalidDelayedState = errors.New("the message is invalid for delayed")
	PubQueue                      = 500
	errChannelNotExist            = errors.New("channel does not exist")
)
func writeMessageToBackend(writeExt bool, buf *bytes.Buffer, msg *Message, bq *diskQueueWriter) (BackendOffset, int32, diskQueueEndInfo, error) {
	buf.Reset()
	_, err := msg.WriteTo(buf, writeExt)
	if err != nil {
		return 0, 0, diskQueueEndInfo{}, err
	}
	return bq.PutV2(buf.Bytes())
}
func writeMessageToBackendWithCheck(writeExt bool, buf *bytes.Buffer, msg *Message, checkSize int64, bq *diskQueueWriter) (BackendOffset, int32, diskQueueEndInfo, error) {
	buf.Reset()
	wsize, err := msg.WriteTo(buf, writeExt)
	if err != nil {
		return 0, 0, diskQueueEndInfo{}, err
	}
	if checkSize > 0 && wsize+4 != checkSize {
		return 0, 0, diskQueueEndInfo{}, fmt.Errorf("message write size mismatch %v vs %v", checkSize, wsize+4)
	}
	return bq.PutV2(buf.Bytes())
}
type MsgIDGenerator interface {
	NextID() uint64
	Reset(uint64)
}
type TopicDynamicConf struct {
	AutoCommit               int32
	RetentionDay             int32
	SyncEvery                int64
	OrderedMulti             bool
	MultiPart                bool
	DisableChannelAutoCreate bool
	Ext                      bool
}
type PubInfo struct {
	Done       chan struct{}
	MsgBody    []byte
	ExtContent ext.IExtContent
	StartPub   time.Time
	Err        error
}
type PubInfoChan chan *PubInfo
type MPubInfo struct {
	Done     chan struct{}
	Msgs     []*Message
	StartPub time.Time
	Err      error
}
type MPubInfoChan chan *MPubInfo
type ChannelMetaInfo struct {
	Name           string `json:"name"`
	Paused         bool   `json:"paused"`
	Skipped        bool   `json:"skipped"`
	ZanTestSkipped bool   `json:"zanTestSkipped"`
}
func (cm *ChannelMetaInfo) IsZanTestSkipepd() bool {
	return cm.ZanTestSkipped
}
type Topic struct {
	sync.Mutex
	tname       string
	fullName    string
	partition   int
	channelMap  map[string]*Channel
	channelLock sync.RWMutex
	backend     *diskQueueWriter
	dataPath    string
	flushChan   chan int
	exitFlag    int32
	ephemeral bool
	deleter   sync.Once
	nsqdNotify      INsqdNotify
	option          *Options
	msgIDCursor     MsgIDGenerator
	defaultIDSeq    uint64
	needFlush       int32
	EnableTrace     int32
	lastSyncCnt     int64
	putBuffer       bytes.Buffer
	bp              sync.Pool
	writeDisabled   int32
	dynamicConf     *TopicDynamicConf
	isOrdered       int32
	magicCode       int64
	committedOffset atomic.Value
	detailStats     *DetailStatsInfo
	needFixData     int32
	pubWaitingChan  PubInfoChan
	mpubWaitingChan MPubInfoChan
	quitChan        chan struct{}
	pubLoopFunc     func(v *Topic)
	wg              sync.WaitGroup
	delayedQueue                 atomic.Value
	isExt                        int32
	isChannelAutoCreatedDisabled int32
	pubFailedCnt                 int64
	metaStorage                  IMetaStorage
	pubWaitingBytes int64
	tpLog           *levellogger.LevelLogger
	kvTopic         *KVTopic
}
func (t *Topic) setExt() {
	atomic.StoreInt32(&t.isExt, 1)
}
func (t *Topic) IsExt() bool {
	return atomic.LoadInt32(&t.isExt) == 1
}
func (t *Topic) DisableChannelAutoCreate() {
	atomic.StoreInt32(&t.isChannelAutoCreatedDisabled, 1)
}
func (t *Topic) EnableChannelAutoCreate() {
	atomic.StoreInt32(&t.isChannelAutoCreatedDisabled, 0)
}
func (t *Topic) IsChannelAutoCreateDisabled() bool {
	return atomic.LoadInt32(&t.isChannelAutoCreatedDisabled) == 1
}
func (t *Topic) IncrPubWaitingBytes(sz int64) int64 {
	return atomic.AddInt64(&t.pubWaitingBytes, sz)
}
func (t *Topic) IncrPubFailed() {
	atomic.AddInt64(&t.pubFailedCnt, 1)
	TopicPubFailedCnt.With(prometheus.Labels{
		"topic":     t.GetTopicName(),
		"partition": strconv.Itoa(t.GetTopicPart()),
	}).Inc()
}
func (t *Topic) PubFailed() int64 {
	return atomic.LoadInt64(&t.pubFailedCnt)
}
func GetTopicFullName(topic string, part int) string {
	return topic + "-" + strconv.Itoa(part)
}
func NewTopicForDelete(topicName string, part int, opt *Options,
	writeDisabled int32, metaStorage IMetaStorage,
	notify INsqdNotify, loopFunc func(v *Topic)) *Topic {
	return NewTopicWithExt(topicName, part, false, false, opt, writeDisabled, metaStorage, nil, notify, loopFunc)
}
func NewTopicWithExt(topicName string, part int, ext bool, ordered bool, opt *Options,
	writeDisabled int32, metaStorage IMetaStorage,
	kvEng engine.KVEngine,
	notify INsqdNotify, loopFunc func(v *Topic)) *Topic {
	return NewTopicWithExtAndDisableChannelAutoCreate(topicName, part, ext, ordered, false, opt, writeDisabled, metaStorage, kvEng, notify, loopFunc)
}
func NewTopicWithExtAndDisableChannelAutoCreate(topicName string, part int, ext bool, ordered bool,
	disbaleChannelAutoCreate bool, opt *Options,
	writeDisabled int32, metaStorage IMetaStorage,
	kvEng engine.KVEngine,
	notify INsqdNotify, loopFunc func(v *Topic)) *Topic {
	if part > MAX_TOPIC_PARTITION {
		return nil
	}
	t := &Topic{
		tname:           topicName,
		partition:       part,
		channelMap:      make(map[string]*Channel),
		flushChan:       make(chan int, 10),
		option:          opt,
		dynamicConf:     &TopicDynamicConf{SyncEvery: opt.SyncEvery, AutoCommit: 1},
		putBuffer:       bytes.Buffer{},
		nsqdNotify:      notify,
		writeDisabled:   writeDisabled,
		pubWaitingChan:  make(PubInfoChan, PubQueue),
		mpubWaitingChan: make(MPubInfoChan, PubQueue/2),
		quitChan:        make(chan struct{}),
		pubLoopFunc:     loopFunc,
		metaStorage:     metaStorage,
		kvTopic:         NewKVTopicWithEngine(topicName, part, ext, kvEng),
	}
	t.fullName = GetTopicFullName(t.tname, t.partition)
	t.tpLog = nsqLog.WrappedWithPrefix("["+t.fullName+"]", 0)
	if metaStorage == nil {
		t.metaStorage = &fileMetaStorage{}
	}
	if ext {
		t.setExt()
	}
	if ordered {
		atomic.StoreInt32(&t.isOrdered, 1)
	}
	if t.dynamicConf.SyncEvery < 1 {
		t.dynamicConf.SyncEvery = 1
	}
	if disbaleChannelAutoCreate {
		t.DisableChannelAutoCreate()
	}
	t.bp.New = func() interface{} {
		return &bytes.Buffer{}
	}
	t.dataPath = path.Join(opt.DataPath, topicName)
	err := os.MkdirAll(t.dataPath, 0755)
	if err != nil {
		t.tpLog.LogErrorf("failed to create directory: %v ", err)
		return nil
	}
	backendName := getBackendName(t.tname, t.partition)
	queue, err := newDiskQueueWriterWithMetaStorage(backendName,
		t.dataPath,
		opt.MaxBytesPerFile,
		int32(minValidMsgLength),
		int32(opt.MaxMsgSize)+minValidMsgLength,
		opt.SyncEvery,
		t.metaStorage,
	)
	if err != nil {
		t.tpLog.LogErrorf("failed to init disk queue: %v ", err)
		if err == ErrNeedFixQueueStart || err == ErrNeedFixQueueEnd {
			t.SetDataFixState(true)
		} else {
			return nil
		}
	}
	t.backend = queue.(*diskQueueWriter)
	if err != ErrNeedFixQueueEnd {
		t.UpdateCommittedOffset(t.backend.GetQueueWriteEnd())
	}
	if _, err := t.tryFixKVTopic(); err != nil {
		t.tpLog.LogErrorf("failed to auto fix kv topic while init: %s", err)
		t.SetDataFixState(true)
	}
	err = t.loadMagicCode()
	if err != nil {
		t.tpLog.LogErrorf("failed to load magic code: %v", err)
		return nil
	}
	t.detailStats = NewDetailStatsInfo(t.tname, strconv.Itoa(t.partition), t.TotalDataSize(), t.getHistoryStatsFileName())
	t.nsqdNotify.NotifyStateChanged(t, true)
	t.tpLog.LogDebugf("new topic created: %v", t.tname)
	if t.pubLoopFunc != nil {
		t.wg.Add(1)
		go func() {
			defer t.wg.Done()
			t.pubLoopFunc(t)
		}()
	}
	t.LoadChannelMeta()
	return t
}
func (t *Topic) GetDelayedQueue() *DelayQueue {
	if t.IsOrdered() {
		return nil
	}
	if t.delayedQueue.Load() == nil {
		return nil
	}
	return t.delayedQueue.Load().(*DelayQueue)
}
func (t *Topic) GetOrCreateDelayedQueueForReadNoLock() (*DelayQueue, error) {
	if t.delayedQueue.Load() == nil {
		delayedQueue, err := NewDelayQueueForRead(t.tname, t.partition, t.dataPath, t.option, nil, t.IsExt())
		if err == nil {
			t.delayedQueue.Store(delayedQueue)
		} else {
			t.tpLog.LogWarningf("init delayed queue error %v", err)
			return nil, err
		}
	}
	return t.delayedQueue.Load().(*DelayQueue), nil
}
func (t *Topic) GetOrCreateDelayedQueueNoLock(idGen MsgIDGenerator) (*DelayQueue, error) {
	if t.delayedQueue.Load() == nil {
		delayedQueue, err := NewDelayQueue(t.tname, t.partition, t.dataPath, t.option, idGen, t.IsExt())
		if err == nil {
			t.delayedQueue.Store(delayedQueue)
			t.channelLock.RLock()
			for _, ch := range t.channelMap {
				ch.SetDelayedQueue(delayedQueue)
			}
			t.channelLock.RUnlock()
		} else {
			t.tpLog.LogWarningf("init delayed queue error %v", err)
			return nil, err
		}
	}
	return t.delayedQueue.Load().(*DelayQueue), nil
}
func (t *Topic) GetWaitChan() PubInfoChan {
	return t.pubWaitingChan
}
func (t *Topic) IsWaitChanFull() bool {
	return len(t.pubWaitingChan) >= cap(t.pubWaitingChan)-1
}
func (t *Topic) IsMWaitChanFull() bool {
	return len(t.mpubWaitingChan) >= cap(t.mpubWaitingChan)-1
}
func (t *Topic) GetMWaitChan() MPubInfoChan {
	return t.mpubWaitingChan
}
func (t *Topic) QuitChan() <-chan struct{} {
	return t.quitChan
}
func (t *Topic) IsDataNeedFix() bool {
	return atomic.LoadInt32(&t.needFixData) == 1
}
func (t *Topic) SetDataFixState(needFix bool) {
	if needFix {
		atomic.StoreInt32(&t.needFixData, 1)
	} else {
		atomic.StoreInt32(&t.needFixData, 0)
	}
}
func (t *Topic) getMagicCodeFileName() string {
	return path.Join(t.dataPath, "magic"+strconv.Itoa(t.partition))
}
func (t *Topic) saveMagicCode() error {
	var f *os.File
	var err error
	fileName := t.getMagicCodeFileName()
	f, err = os.OpenFile(fileName, os.O_RDWR|os.O_CREATE, 0644)
	if err != nil {
		return err
	}
	defer f.Close()
	_, err = fmt.Fprintf(f, "%d\n",
		atomic.LoadInt64(&t.magicCode))
	if err != nil {
		return err
	}
	f.Sync()
	t.tpLog.Infof("saved as magic code: %v", atomic.LoadInt64(&t.magicCode))
	return nil
}
func (t *Topic) removeMagicCode() {
	fileName := t.getMagicCodeFileName()
	err := os.Remove(fileName)
	if err != nil {
		t.tpLog.Infof("remove the magic file %v failed:%v", fileName, err)
	}
}
func (t *Topic) loadMagicCode() error {
	var f *os.File
	var err error
	fileName := t.getMagicCodeFileName()
	f, err = os.OpenFile(fileName, os.O_RDONLY, 0644)
	if err != nil {
		if os.IsNotExist(err) {
			return nil
		}
		return err
	}
	defer f.Close()
	var code int64
	_, err = fmt.Fscanf(f, "%d\n",
		&code)
	if err != nil {
		return err
	}
	atomic.StoreInt64(&t.magicCode, code)
	t.tpLog.Infof("loading topic as magic code: %v", code)
	return nil
}
func (t *Topic) removeHistoryStat() {
	fileName := t.getHistoryStatsFileName()
	err := os.Remove(fileName)
	if err != nil {
		t.tpLog.Infof("remove file %v failed:%v", fileName, err)
	}
}
func (t *Topic) MarkAsRemoved() (string, error) {
	t.Lock()
	defer t.Unlock()
	atomic.CompareAndSwapInt32(&t.exitFlag, 0, 1)
	t.tpLog.Logf("TOPIC(%s): deleting", t.GetFullName())
	t.nsqdNotify.NotifyStateChanged(t, true)
	t.channelLock.Lock()
	for _, channel := range t.channelMap {
		t.channelMap[channel.name] = nil
		delete(t.channelMap, channel.name)
		channel.Delete()
	}
	t.channelLock.Unlock()
	renamePath := t.dataPath + "-removed-" + strconv.Itoa(int(time.Now().Unix()))
	t.tpLog.Warningf("mark the topic as removed: %v", renamePath)
	os.MkdirAll(renamePath, 0755)
	err := t.backend.RemoveTo(renamePath)
	if err != nil {
		t.tpLog.Errorf("failed to mark the topic as removed %v failed: %v", renamePath, err)
	}
	util.AtomicRename(t.getMagicCodeFileName(), path.Join(renamePath, "magic"+strconv.Itoa(t.partition)))
	t.removeHistoryStat()
	t.RemoveChannelMeta()
	t.removeMagicCode()
	if t.GetDelayedQueue() != nil {
		t.GetDelayedQueue().Delete()
	}
	return renamePath, err
}
func (t *Topic) SetMagicCode(code int64) error {
	if t.magicCode == code {
		return nil
	} else if t.magicCode > 0 {
		t.tpLog.Errorf("magic code set more than once :%v, %v", t.magicCode, code)
		return errors.New("magic code can not be changed.")
	}
	t.magicCode = code
	return t.saveMagicCode()
}