TopicDiff { public static TopicDiff diff(Topic source, Topic target) { Objects.requireNonNull(source); Objects.requireNonNull(target); Objects.requireNonNull(source.getTopicName()); Objects.requireNonNull(target.getTopicName()); if (!source.getTopicName().equals(target.getTopicName())) { throw new IllegalArgumentException(); } Map<String, Difference> differences = new HashMap<>(); if (source.getNumPartitions() != target.getNumPartitions()) { NumPartitionsDifference numPartitionsDifference = new NumPartitionsDifference(source.getNumPartitions(), target.getNumPartitions()); differences.put(numPartitionsDifference.address(), numPartitionsDifference); } if (source.getNumReplicas() != target.getNumReplicas()) { NumReplicasDifference numReplicasDifference = new NumReplicasDifference(target.getNumReplicas()); differences.put(numReplicasDifference.address(), numReplicasDifference); } if (!source.getConfig().equals(target.getConfig())) { HashSet<String> removed = new HashSet<>(source.getConfig().keySet()); removed.removeAll(target.getConfig().keySet()); for (String removedKey : removed) { RemovedConfigEntry removedConfigEntry = new RemovedConfigEntry(removedKey); differences.put(removedConfigEntry.address(), removedConfigEntry); } HashSet<String> added = new HashSet<>(target.getConfig().keySet()); added.removeAll(source.getConfig().keySet()); for (String addedKey : added) { AddedConfigEntry addedConfigEntry = new AddedConfigEntry(addedKey, target.getConfig().get(addedKey)); differences.put(addedConfigEntry.address(), addedConfigEntry); } HashSet<String> retained = new HashSet<>(source.getConfig().keySet()); retained.retainAll(target.getConfig().keySet()); for (String retainedKey : retained) { if (!source.getConfig().get(retainedKey).equals(target.getConfig().get(retainedKey))) { AddedConfigEntry addedConfigEntry = new AddedConfigEntry(retainedKey, target.getConfig().get(retainedKey)); differences.put(addedConfigEntry.address(), addedConfigEntry); } } } return new TopicDiff(differences, target.getMetadata()); } private  TopicDiff(Map<String, Difference> differences, ObjectMeta objectMeta); static TopicDiff diff(Topic source, Topic target); @Override boolean equals(Object o); @Override int hashCode(); @Override String toString(); boolean isEmpty(); boolean changesNumPartitions(); int numPartitionsDelta(); boolean changesConfig(); boolean changesReplicationFactor(); Topic apply(Topic topic); String conflict(TopicDiff other); boolean conflicts(TopicDiff other); TopicDiff merge(TopicDiff other);  }
@Test public void testDiffDifferentTopicNamesThrows() { Topic topicBWrongName = new Topic.Builder("another_name", 3, topicBConfig).build(); assertThrows(IllegalArgumentException.class, () -> TopicDiff.diff(topicA, topicBWrongName)); }