private static List<String> getGroupsForTopic(String brokerServers, String topic)

             throws ExecutionException, InterruptedException, TimeoutException {

         Properties props = new Properties();

         props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerServers);

         try (AdminClient client = AdminClient.create(props)) {

             List<String> allGroups = client.listConsumerGroups()

                     .valid()

                     .get(10, TimeUnit.SECONDS)

                     .stream()

                     .map(ConsumerGroupListing::groupId)

                     .collect(Collectors.toList());

             Map<String, ConsumerGroupDescription> allGroupDetails =

                     client.describeConsumerGroups(allGroups).all().get(10, TimeUnit.SECONDS);

             final List<String> filteredGroups = new ArrayList<>();

             allGroupDetails.entrySet().forEach(entry -> {

                 String groupId = entry.getKey();

                 ConsumerGroupDescription description = entry.getValue();

                 boolean topicSubscribed = description.members().stream().map(MemberDescription::assignment)

                         .map(MemberAssignment::topicPartitions)

                         .map(tps -> tps.stream().map(TopicPartition::topic).collect(Collectors.toSet()))

                         .anyMatch(tps -> tps.contains(topic));

                 if (topicSubscribed)

                     filteredGroups.add(groupId);

             });

             return filteredGroups;

         }

     }