[tests] Issue 2639: NullPointerException at closing consumer
* [tests] Issue 2639: NullPointerException at closing consumer
*Motivation*
Connection can be null when closing consumer.
*Changes*
Check if the connection is null or not during closing.
* Address comments
@@ -689,13 +689,28 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
acknowledgmentsGroupingTracker.close();
long requestId = client.newRequestId();
- ByteBuf cmd = Commands.newCloseConsumer(consumerId, requestId);
CompletableFuture<Void> closeFuture = new CompletableFuture<>();
ClientCnx cnx = cnx();
+ if (null == cnx) {
+ cleanupAtClose(closeFuture);
+ } else {
+ ByteBuf cmd = Commands.newCloseConsumer(consumerId, requestId);
cnx.sendRequestWithId(cmd, requestId).handle((v, exception) -> {
cnx.removeConsumer(consumerId);
if (exception == null || !cnx.ctx().channel().isActive()) {
+ cleanupAtClose(closeFuture);
+ } else {
+ closeFuture.completeExceptionally(exception);
+ }
+ return null;
+ });
+ }
+
+ return closeFuture;
+ }
+
+ private void cleanupAtClose(CompletableFuture<Void> closeFuture) {
log.info("[{}] [{}] Closed consumer", topic, subscription);
setState(State.Closed);
unAckedMessageTracker.close();
@@ -706,13 +721,6 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
client.cleanupConsumer(this);
// fail all pending-receive futures to notify application
failPendingReceive();
- } else {
- closeFuture.completeExceptionally(exception);
- }
- return null;
- });
-
- return closeFuture;
}
private void failPendingReceive() {
