[GitHub] tfhappy opened a new issue #3480: ack not return error when consumer closed

[GitBox]

tfhappy opened a new issue #3480: ack not return error when consumer closed
URL: https://github.com/apache/pulsar/issues/3480
 
 
   **Describe the bug**
   just like the title
   
   **To Reproduce**
   Steps to reproduce the behavior:
   1. Send 10 or more msg
   2. Run code
   3. You can still receive a message when consumer closes（ReceiverQueueSize = 
-1 invalid？）
       and `consumer.Ack(msg)` not return an error
   4. Run code again to validate
   
   cunsumer code:
   ```
   package main
   
   import (
        "log"
        "runtime"
   
        "fmt"
        "github.com/apache/pulsar/pulsar-client-go/pulsar"
        "time"
   )
   
   func main() {
        // Creating a client
        client, err := pulsar.NewClient(pulsar.ClientOptions{
                URL: "pulsar://localhost:6650",
                OperationTimeoutSeconds: 5,
                MessageListenerThreads:  runtime.NumCPU(),
        })
   
        if err != nil {
                log.Fatalf("Could not instantiate Pulsar client: %v", err)
        }
   
        // Creating a consumer
        msgChannel := make(chan pulsar.ConsumerMessage)
   
        consumerOpts := pulsar.ConsumerOptions{
                Topic:             "my-topic",
                SubscriptionName:  "my-subscription-1",
                Type:              pulsar.Exclusive,
                MessageChannel:    msgChannel,
                ReceiverQueueSize: -1,
        }
   
        consumer, err := client.Subscribe(consumerOpts)
   
        if err != nil {
                log.Fatalf("Could not establish subscription: %v", err)
        }
   
        go func() {
                time.Sleep(time.Second * 5)
                consumer.Close()
        }()
   
        for cm := range msgChannel {
                msg := cm.Message
                err := consumer.Ack(msg)
   
                fmt.Printf("Message value: %s; err: %v\n", 
string(msg.Payload()), err)
   
                time.Sleep(time.Second)
        }
   }
   
   ```
   Result:
   ```
   # First run
   Message value: Hello, Pulsar 0; err: <nil>
   Message value: Hello, Pulsar 1; err: <nil>
   Message value: Hello, Pulsar 2; err: <nil>
   Message value: Hello, Pulsar 3; err: <nil>
   Message value: Hello, Pulsar 4; err: <nil>
   2019/01/30 17:55:08 INFO  | ConsumerImpl:761 | 
[persistent://public/default/my-topic, my-subscription-1, 0] Closed consumer 0
   Message value: Hello, Pulsar 5; err: <nil>
   
   # Second run
   Message value: Hello, Pulsar 5; err: <nil>
   Message value: Hello, Pulsar 6; err: <nil>
   ...
   ```
   **Expected behavior**
   `consumer.Ack(msg)` return an error but not nil when consumer closed
   
   **Desktop**
   - OS: ubuntu18.04
   - version: 2.2.0
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services