[Issue 1218][Reader] Reader Next returns on closed consumer

pulsar/reader_impl.go

@@ -171,6 +171,8 @@ func (r *reader) Next(ctx context.Context) (Message, error) {
 				return nil, err
 			}
 			return cm.Message, nil
+		case <-r.c.closeCh:
+			return nil, newError(ConsumerClosed, "consumer closed")
 		case <-ctx.Done():
 			return nil, ctx.Err()
 		}

pulsar/reader_test.go

@@ -1035,3 +1035,27 @@ func TestReaderHasNextRetryFailed(t *testing.T) {
 	}
 
 }
+
+func TestReaderNextReturnsOnClosedConsumer(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL:              serviceURL,
+		OperationTimeout: 2 * time.Second,
+	})
+	assert.NoError(t, err)
+	topic := newTopicName()
+	reader, err := client.CreateReader(ReaderOptions{
+		Topic:          topic,
+		StartMessageID: EarliestMessageID(),
+	})
+	assert.Nil(t, err)
+
+	reader.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	var e *Error
+	_, err = reader.Next(ctx)
+	assert.ErrorAs(t, err, &e)
+	assert.Equal(t, ConsumerClosed, e.Result())
+}

pulsar/table_view_impl.go

@@ -129,7 +129,9 @@ func (tv *TableViewImpl) partitionUpdateCheck() error {
 				if err != nil {
 					tv.logger.Errorf("read next message failed for %s: %w", partition, err)
 				}
-				tv.handleMessage(msg)
+				if msg != nil {
+					tv.handleMessage(msg)
+				}
 			}
 			ctx, cancelFunc := context.WithCancel(context.Background())
 			tv.cancelRaders[partition] = cancelReader{
@@ -268,9 +270,12 @@ func (tv *TableViewImpl) watchReaderForNewMessages(ctx context.Context, reader R
 		if err != nil {
 			tv.logger.Errorf("read next message failed for %s: %w", reader.Topic(), err)
 		}
-		if errors.Is(err, context.Canceled) {
+		var e *Error
+		if (errors.As(err, &e) && e.Result() == ConsumerClosed) || errors.Is(err, context.Canceled) {
 			return
 		}
-		tv.handleMessage(msg)
+		if msg != nil {
+			tv.handleMessage(msg)
+		}
 	}
 }