internal/queue/kafka/hub.go에서 panic 발생

[cg10036]

Hub Shutdown과 동시에 해당 함수가 실행된다면 panic이 발생합니다.

queue-manager/internal/queue/kafka/hub.go

Lines 99 to 106 in b8184d4

	func (h *Hub) GetConsumer(consumerId int) *Consumer {
	mutex := sync.Mutex{}
	mutex.Lock()
	defer mutex.Unlock()
	return h.consumers[consumerId]
	}

queue-manager/internal/queue/kafka/hub.go

Lines 108 to 113 in b8184d4

	func (h *Hub) GetRandomConsumer() *Consumer {
	mutex := sync.Mutex{}
	mutex.Lock()
	consumer := h.consumers[0]
	mutex.Unlock()

queue-manager/internal/queue/kafka/hub.go

Lines 139 to 145 in b8184d4

	func (h *Hub) MakeNewConsumer(ctx *context.Context) *Consumer {
	mutex := sync.Mutex{}
	defer mutex.Unlock()
	consumer := NewConsumer(h.availID, h.messageSerializer, h.publishOnly, h.config)
	mutex.Lock()

queue-manager/internal/queue/kafka/hub.go

Lines 167 to 180 in b8184d4

	func (h *Hub) SendMessage(
	item queueitem.Universal,
	topic string,
	consumerId int,
	) error {
	if !h.isRunning {
	return queueerror.ErrQueueNotPrepared(topic)
	}
	mutex := sync.Mutex{}
	defer mutex.Unlock()
	mutex.Lock()

queue-manager/internal/queue/kafka/hub.go

Lines 187 to 212 in b8184d4

	func (h *Hub) Shutdown() {
	if !h.isRunning {
	return
	}
	// Shutdown waits until all consumers are stopped.
	mutex := sync.Mutex{}
	shutdownWaitGroup := sync.WaitGroup{}
	copyedConsumers := make(map[int]*Consumer, h.consumersMax) // To prevent deadlock while iterating over consumers
	for consumerId, consumer := range h.consumers {
	copyedConsumers[consumerId] = consumer
	}
	shutdownWaitGroup.Add(1)
	go func() {
	defer shutdownWaitGroup.Done()
	for consumerId, consumer := range copyedConsumers {
	consumer.Pause()
	mutex.Lock()
	delete(h.consumers, consumerId)
	mutex.Unlock()
	}
	}()

Hub Shutdown은 함수 내부에서 뮤텍스를 생성해 사용하기 때문에 위에 언급된 함수가 사용하는 mutex와 다릅니다. 따라서 Shutdown 도중 위에 언급된 함수가 실행되면 map에서 동시 Read/Write를 하게 되고 panic이 발생합니다. 만약 동시 실행이 절대 불가능한 환경이라면 mutex를 제거하는 것이 좋아보입니다.