private static <T extends Map<String, String>> T collectNonEmptyTags(
        Map<String, String> tags,
        Supplier<T> mapSupplier
    ) {
        T result = mapSupplier.get();
        for (Map.Entry<String, String> tagEntry : tags.entrySet()) {
            String tagValue = tagEntry.getValue();
            if (!"".equals(tagValue)) {
                result.put(tagEntry.getKey(), tagValue);
            }
        }
        return result;
    }
convenience method to replace {@link com.yammer.metrics.Metrics#defaultRegistry()}
server-common/src/main/java/org/apache/kafka/server/metrics/KafkaYammerMetrics.java