func consumer3() {
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL:    "pulsar://localhost:6650",
		Logger: plog.NewLoggerWithLogrus(&logrus.Logger{}),
	})
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	schema := pulsar.NewAvroSchema(common.AvroSchema, nil)

	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
		Topic:                       "test1",
		SubscriptionName:            "my-shared-subscription",
		Type:                        pulsar.Shared,
		Schema:                      schema,
		SubscriptionInitialPosition: pulsar.SubscriptionPositionEarliest,
		EnableZeroQueueConsumer:     true,
		MaxPendingChunkedMessage:    1,
	})

	if err != nil {
		log.Fatal(err)
	}
	defer consumer.Close()

	accountWorker := common.NewAccountListener(5)

	fmt.Println("waiting..")
	for accountWorker.WaitReady() {
		msg, err := consumer.Receive(context.Background())
		if err != nil {
			log.Printf("Error receiving message: %v", err)
			continue
		}
		account := &common.Account{}
		err = msg.GetSchemaValue(account)
		if err != nil {
			log.Printf("Error decoding message: %v", err)
			consumer.Nack(msg)
			continue
		}
		fmt.Println("Consumed : " + strconv.Itoa(account.ID))
		accountWorker.AddAccount(consumer, msg, account)
	}
} 

const AvroSchema = `
{
  "type": "record",
  "name": "Account",
  "fields": [
    {"name": "ID", "type": "int"}
  ]
}`

type AccountWrapper struct {
	Account  *Account
	Consumer pulsar.Consumer
	Message  pulsar.Message
}

type Account struct {
	ID int `avro:"ID"`
}

type AccountListener struct {
	Limit    int
	Accounts map[int]*AccountWrapper
	lock     *sync.Mutex
}

func NewAccountListener(limit int) *AccountListener {
	return &AccountListener{
		Limit:    limit,
		Accounts: make(map[int]*AccountWrapper),
		lock:     &sync.Mutex{},
	}
}

func (l *AccountListener) WaitReady() bool {
	for l.checkLimit() == false {
		time.Sleep(time.Millisecond * 100)
	}
	return true
}

func (l *AccountListener) IsReady() bool {
	l.lock.Lock()
	defer l.lock.Unlock()
	return len(l.Accounts) < l.Limit
}

func (l *AccountListener) checkLimit() bool {
	l.lock.Lock()
	defer l.lock.Unlock()
	return len(l.Accounts) < l.Limit
}

func (l *AccountListener) AddAccount(consumer pulsar.Consumer, message pulsar.Message, account *Account) {
	l.lock.Lock()
	defer l.lock.Unlock()
	l.Accounts[account.ID] = &AccountWrapper{
		Account:  account,
		Message:  message,
		Consumer: consumer,
	}
	go l.Process(l.Accounts[account.ID])
}

func (l *AccountListener) RemoveAccount(account *Account) {
	l.lock.Lock()
	defer l.lock.Unlock()
	delete(l.Accounts, account.ID)
}

func (l *AccountListener) Process(wrapper *AccountWrapper) {
	// long process simulate
	time.Sleep(time.Minute * 5)
	// finish
	fmt.Println("< : " + strconv.Itoa(wrapper.Account.ID))
	wrapper.Consumer.Ack(wrapper.Message) // success - remove queue
	l.RemoveAccount(wrapper.Account)
}

if pc.options.receiverQueueSize > 0 {
    pc.availablePermits.inc()
}