KafkaBrokerMonitor { @Scheduled(fixedDelayString = "${kafka.broker.monitor.fixed.delay:60000}") public void checkAndUpdateBrokers() throws KafkaException { adminClient.getBrokerIds().forEach(this::checkAndUpdateBroker); } @Autowired  KafkaBrokerMonitor(
      ZkUtils zkUtils,
      // 52428800 = 50MiB
      @Value("${kafka.throttle.leader.rate:52428800}") String leaderThrottledRateBytes,
      @Value("${kafka.throttle.follower.rate:52428800}") String followerThrottledRateBytes); @Scheduled(fixedDelayString = "${kafka.broker.monitor.fixed.delay:60000}") void checkAndUpdateBrokers();  }
@Test public void testName() throws Exception { doReturn(Collections.singletonList("0")).when(client).getBrokerIds(); doReturn(new Properties()).when(client).getConfig("0"); underTest.checkAndUpdateBrokers(); ArgumentCaptor<Properties> captor = ArgumentCaptor.forClass(Properties.class); verify(client).changeConfig(eq("0"), captor.capture()); Properties config = captor.getValue(); assertThat(config.size(), is(2)); assertThat(config.getProperty("leader.replication.throttled.rate"), is("1")); assertThat(config.getProperty("follower.replication.throttled.rate"), is("2")); }