/**

     * get all subscribing consumer group names for a given topic

     * @param brokerListUrl localhost:9092 for instance

     * @param topic         topic name

     * @return

     */

    public static Set<String> getAllGroupsForTopic(String brokerListUrl, String topic) {

        AdminClient client = AdminClient.createSimplePlaintext(brokerListUrl);

        try {

            List<GroupOverview> allGroups = scala.collection.JavaConversions.seqAsJavaList(client.listAllGroupsFlattened().toSeq());

            Set<String> groups = new HashSet<>();

            for (GroupOverview overview: allGroups) {

                String groupID = overview.groupId();

                Map<TopicPartition, Object> offsets = scala.collection.JavaConversions.mapAsJavaMap(client.listGroupOffsets(groupID));

                Set<TopicPartition> partitions = offsets.keySet();

                for (TopicPartition tp: partitions) {

                    if (tp.topic().equals(topic)) {

                        groups.add(groupID);

                    }

                }

            }

            return groups;

        } finally {

            client.close();

        }

    }