Add test to verify that cleanup.policy is set to compact in the KafkaTopicClient

[apurvam]

No description provided.

[apurvam]

This is a follow up to #1042

[hjafarpour]

The order of topics may not be the same in the two collections.

[apurvam]

Yes. But for now, we only are testing with lists of single size, since our API creates one topic at a time. So it isn't an issue IMO.

[hjafarpour]

Should these be != instead of ==?

[apurvam]

No. The ! is over the whole condition. So everything has to be equal.

[big-andy-coates]

Hey @apurvam - I've suggested a cleaner way of testing this than a custom comparator. See what you think.

[big-andy-coates]

private static class, and could we move this to the bottom of the file please?

[big-andy-coates]

That's a pretty hacky comparator impl! ;)

Comparators are all about sorting collections, which is not what you want to achieve here. Why not just use a custom argument matcher that does what you want? e.g.

private static Collection<NewTopic> singleNewTopic(final NewTopic expected) {
    class NewTopicsMatcher implements IArgumentMatcher {
      @SuppressWarnings("unchecked")
      @Override
      public boolean matches(final Object argument) {
        final Collection<NewTopic> newTopics = (Collection<NewTopic>) argument;
        if (newTopics.size() != 1) {
          return false;
        }

        final NewTopic actual = newTopics.iterator().next();
        return Objects.equals(actual.name(), expected.name())
               && Objects.equals(actual.replicationFactor(), expected.replicationFactor())
               && Objects.equals(actual.numPartitions(), expected.numPartitions())
               && Objects.equals(actual.configs(), expected.configs());
      }

      @Override
      public void appendTo(final StringBuffer buffer) {
        buffer.append("{NewTopic").append(expected).append("}");
      }
    }

    EasyMock.reportMatcher(new NewTopicsMatcher());
    return null;
  }

Then you can change the test to:

  @Test
  public void shouldSetTopicCleanupPolicyToCompact()  {
    AdminClient adminClient = mock(AdminClient.class);
    expect(adminClient.describeCluster()).andReturn(getDescribeClusterResult());
    expect(adminClient.listTopics()).andReturn(getEmptyListTopicResult());
    expect(adminClient.describeConfigs(anyObject())).andReturn(getDescribeConfigsResult());

    // Verify that the new topic configuration being passed to the admin client is what we expect.
    NewTopic newTopic = new NewTopic(topicName1, 1, (short) 1);
    newTopic.configs(Collections.singletonMap("cleanup.policy", "compact"));
  
    expect(adminClient.createTopics(singleNewTopic(newTopic))).andReturn(getCreateTopicsResult());
    replay(adminClient);
    
    KafkaTopicClient kafkaTopicClient = new KafkaTopicClientImpl(adminClient);
    kafkaTopicClient.createTopic(topicName1, 1, (short) 1, true);
    verify(adminClient);
  }

[apurvam]

Thanks @big-andy-coates . One of the motivations for doing this was to learn about how to do this nicely with EasyMock. Thanks for the tip! I implemented your changes.

[hjafarpour]

LGTM.

[rodesai]

LGTM!