LeaderInSyncByPartitionFunction { public Map<TopicPartition, LeaderInSync> apply(int brokerId, Collection<String> topics) { return Mono .fromSupplier(() -> client.describeTopics(topics).all()) .map(KafkaFutures::join) .flatMapIterable(Map::values) .flatMap(td -> Flux .fromIterable(td.partitions()) .filter(tpi -> Flux.fromIterable(tpi.replicas()).any(n -> n.id() == brokerId).block()) .map(tpi -> { TopicPartition partition = new TopicPartition(td.name(), tpi.partition()); boolean leader = tpi.leader() == null ? false : tpi.leader().id() == brokerId; boolean inSync = Flux.fromIterable(tpi.isr()).any(n -> n.id() == brokerId).block(); return Tuples.of(partition, new LeaderInSync(leader, inSync)); })) .collectMap(Tuple2::getT1, Tuple2::getT2) .block(); }  Map<TopicPartition, LeaderInSync> apply(int brokerId, Collection<String> topics);  }
@Test public void typical() throws Exception { TopicPartitionInfo tpi = new TopicPartitionInfo(0, node0, singletonList(node0), singletonList(node0)); KafkaFuture<Map<String, TopicDescription>> kafkaFuture = topicDescriptionFuture(tpi); doReturn(describeTopicsResult).when(adminClient).describeTopics(topics); doReturn(kafkaFuture).when(describeTopicsResult).all(); Map<TopicPartition, LeaderInSync> result = underTest.apply(0, topics); assertThat(result.size(), is(1)); LeaderInSync leaderInSync = result.get(new TopicPartition(topic, 0)); assertThat(leaderInSync.isLeader(), is(true)); assertThat(leaderInSync.isInSync(), is(true)); }
@Test public void notLocalBroker() throws Exception { TopicPartitionInfo tpi = new TopicPartitionInfo(0, node0, singletonList(node1), singletonList(node0)); KafkaFuture<Map<String, TopicDescription>> kafkaFuture = topicDescriptionFuture(tpi); doReturn(describeTopicsResult).when(adminClient).describeTopics(topics); doReturn(kafkaFuture).when(describeTopicsResult).all(); Map<TopicPartition, LeaderInSync> result = underTest.apply(0, topics); assertThat(result.size(), is(0)); }
@Test public void notLeader() throws Exception { TopicPartitionInfo tpi = new TopicPartitionInfo(0, node1, singletonList(node0), singletonList(node0)); KafkaFuture<Map<String, TopicDescription>> kafkaFuture = topicDescriptionFuture(tpi); doReturn(describeTopicsResult).when(adminClient).describeTopics(topics); doReturn(kafkaFuture).when(describeTopicsResult).all(); Map<TopicPartition, LeaderInSync> result = underTest.apply(0, topics); assertThat(result.size(), is(1)); LeaderInSync leaderInSync = result.get(new TopicPartition(topic, 0)); assertThat(leaderInSync.isLeader(), is(false)); assertThat(leaderInSync.isInSync(), is(true)); }
@Test public void notInSync() throws Exception { TopicPartitionInfo tpi = new TopicPartitionInfo(0, node0, singletonList(node0), singletonList(node1)); KafkaFuture<Map<String, TopicDescription>> kafkaFuture = topicDescriptionFuture(tpi); doReturn(describeTopicsResult).when(adminClient).describeTopics(topics); doReturn(kafkaFuture).when(describeTopicsResult).all(); Map<TopicPartition, LeaderInSync> result = underTest.apply(0, topics); assertThat(result.size(), is(1)); LeaderInSync leaderInSync = result.get(new TopicPartition(topic, 0)); assertThat(leaderInSync.isLeader(), is(true)); assertThat(leaderInSync.isInSync(), is(false)); }
@Test public void noLeader() throws Exception { TopicPartitionInfo tpi = new TopicPartitionInfo(0, null, singletonList(node0), singletonList(node0)); KafkaFuture<Map<String, TopicDescription>> kafkaFuture = topicDescriptionFuture(tpi); doReturn(describeTopicsResult).when(adminClient).describeTopics(topics); doReturn(kafkaFuture).when(describeTopicsResult).all(); Map<TopicPartition, LeaderInSync> result = underTest.apply(0, topics); assertThat(result.size(), is(1)); LeaderInSync leaderInSync = result.get(new TopicPartition(topic, 0)); assertThat(leaderInSync.isLeader(), is(false)); assertThat(leaderInSync.isInSync(), is(true)); }