Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java (Java)

Code before:
            // enough that there aren't any outstanding requests to be sent anyway.
            if (networkClientDelegate != null)
                sendUnsentRequests(timer);

            asyncConsumerMetrics.recordApplicationEventExpiredSize(applicationEventReaper.reap(applicationEventQueue));

            closeQuietly(requestManagers, "request managers");
            closeQuietly(networkClientDelegate, "network client delegate");
            log.debug("Closed the consumer network thread");
        }
    }

    /**
     * If there is a metadata error, complete all uncompleted events that require subscription metadata.
     */
    private void maybeFailOnMetadataError(List<CompletableEvent<?>> events) {
        List<CompletableApplicationEvent<?>> subscriptionMetadataEvent = new ArrayList<>();

        for (CompletableEvent<?> ce : events) {
            if (!(ce instanceof CompletableApplicationEvent))
                continue;

            CompletableApplicationEvent<?> cae = (CompletableApplicationEvent<?>) ce;

            if (cae.requireSubscriptionMetadata())
                subscriptionMetadataEvent.add(cae);
        }

        if (subscriptionMetadataEvent.isEmpty())
            return;
        networkClientDelegate.getAndClearMetadataError().ifPresent(metadataError ->
                subscriptionMetadataEvent.forEach(event -> event.future().completeExceptionally(metadataError))
        );
    }
}


Review comment (refactor):
maybe simplify ?
```suggestion
            if (ce instanceof CompletableApplicationEvent && ((CompletableApplicationEvent<?>) ce).requireSubscriptionMetadata()) {
                subscriptionMetadataEvent.add((CompletableApplicationEvent<?>) ce);
            }
```

Code after:
            // to be null, so check before using. If the network client is null, things have failed catastrophically
            // enough that there aren't any outstanding requests to be sent anyway.
            if (networkClientDelegate != null)
                sendUnsentRequests(timer);

            asyncConsumerMetrics.recordApplicationEventExpiredSize(applicationEventReaper.reap(applicationEventQueue));

            closeQuietly(requestManagers, "request managers");
            closeQuietly(networkClientDelegate, "network client delegate");
            log.debug("Closed the consumer network thread");
        }
    }

    /**
     * If there is a metadata error, complete all uncompleted events that require subscription metadata.
     */
    private void maybeFailOnMetadataError(List<CompletableEvent<?>> events) {
        List<CompletableApplicationEvent<?>> subscriptionMetadataEvent = new ArrayList<>();

        for (CompletableEvent<?> ce : events) {
            if (ce instanceof CompletableApplicationEvent && ((CompletableApplicationEvent<?>) ce).requireSubscriptionMetadata())
                subscriptionMetadataEvent.add((CompletableApplicationEvent<?>) ce);
        }

        if (subscriptionMetadataEvent.isEmpty())
            return;
        networkClientDelegate.getAndClearMetadataError().ifPresent(metadataError ->
                subscriptionMetadataEvent.forEach(event -> event.future().completeExceptionally(metadataError))
        );
    }
}