### Question:
TopicEnforcer extends Enforcer<ConfiguredTopic> { List<ConfiguredTopic> increasePartitions() { List<ConfiguredTopic> toIncrease = topicsWithConfigDrift(Type.PARTITION_COUNT, configDriftSafetyFilters, true); if (!toIncrease.isEmpty()) { LOG.info("Increasing partitions for {} topics: {}", toIncrease.size(), toIncrease); topicService.increasePartitions(toIncrease); } else { LOG.info("No partition count drift was detected."); } return toIncrease; } TopicEnforcer(
      TopicService topicService,
      List<ConfiguredTopic> configuredTopics,
      ConfigDrift configDrift,
      boolean safemode);  TopicEnforcer(
      TopicService topicService, List<ConfiguredTopic> configuredTopics, boolean safemode); @Override void stats();  }### Answer:
@Test public void testPartitionIncrease() { List<ConfiguredTopic> configured = Arrays.asList( new ConfiguredTopic("a", 1, (short) 3, Collections.emptyMap()), new ConfiguredTopic("b", 3, (short) 1, Collections.emptyMap()), new ConfiguredTopic("c", 1, (short) 1, Collections.emptyMap())); Map<String, ConfiguredTopic> existing = new HashMap<String, ConfiguredTopic>() { { put("a", new ConfiguredTopic("a", 1, (short) 1, Collections.emptyMap())); put("b", new ConfiguredTopic("b", 1, (short) 1, Collections.emptyMap())); put("d", new ConfiguredTopic("b", 1, (short) 1, Collections.emptyMap())); } }; when(service.listExisting(true)).thenReturn(existing); TopicEnforcer enforcer = new TopicEnforcer(service, configured, true); enforcer.increasePartitions(); List<ConfiguredTopic> expected = Collections.singletonList(configured.get(1)); verify(service).listExisting(true); verify(service).increasePartitions(expected); verifyNoMoreInteractions(service); }
@Test public void testPartitionIncreaseUnSafeMode() { List<ConfiguredTopic> configured = Collections.singletonList( new ConfiguredTopic("a", 1000, (short) 3, Collections.emptyMap())); Map<String, ConfiguredTopic> existing = Collections.singletonMap("a", new ConfiguredTopic("a", 100, (short) 3, Collections.emptyMap())); when(service.listExisting(true)).thenReturn(existing); enforcer = new TopicEnforcer(service, configured, false); Assert.assertEquals("aggressive partition count increase must be allowed in unsafe mode", configured, enforcer.increasePartitions()); }