FlatBinaryMaxRateRegistry implements MaxRateRegistry, NodeCacheListener { @Override public void writeRateHistory(ConsumerInstance consumer, RateHistory rateHistory) { Preconditions.checkState(consumer.getConsumerId().equals(consumerId), "Saving rate history is allowed only for current consumer"); currentConsumerRateHistories.setRateHistory(consumer.getSubscription(), rateHistory); } FlatBinaryMaxRateRegistry(ConfigFactory configFactory,
                              ClusterAssignmentCache clusterAssignmentCache,
                              ConsumerAssignmentCache consumerAssignmentCache,
                              CuratorFramework curator,
                              ZookeeperPaths zookeeperPaths,
                              SubscriptionIds subscriptionIds); @Override void start(); @Override void stop(); @Override void onBeforeMaxRateCalculation(); @Override void onAfterMaxRateCalculation(); @Override Set<ConsumerRateInfo> ensureCorrectAssignments(SubscriptionName subscriptionName, Set<String> currentConsumers); @Override void update(SubscriptionName subscriptionName, Map<String, MaxRate> newMaxRates); @Override Optional<MaxRate> getMaxRate(ConsumerInstance consumer); @Override RateHistory getRateHistory(ConsumerInstance consumer); @Override void writeRateHistory(ConsumerInstance consumer, RateHistory rateHistory); @Override void onAfterWriteRateHistories(); @Override void nodeChanged();  }
@Test(expected = IllegalStateException.class) public void shouldNotAllowForSavingRateHistoryOfOtherThanCurrentConsumerNode() { maxRateRegistry.writeRateHistory(new ConsumerInstance("otherConsumer", subscription1), RateHistory.create(0.5)); }