@Test
@Transactional
public void test_listTopics_withSearchString() throws Exception {

    // Create a cluster.
    final Cluster cluster = clusterTestTools.createCluster(
        "Test Cluster",
        sharedKafkaTestResource.getKafkaConnectString()
    );
    final String searchStr = "CaT";
    // Create some topics
    final String expectedTopic1 = "ExpectedTopic1-" + System.currentTimeMillis();
    final String expectedTopic2= "ExpectedCATTopic2-" + System.currentTimeMillis();
    final String expectedTopic3 = "ExpectedTopic3-" + System.currentTimeMillis();
    sharedKafkaTestResource
        .getKafkaTestUtils()
        .createTopic(expectedTopic1, 1, (short) 1);
    sharedKafkaTestResource
        .getKafkaTestUtils()
        .createTopic(expectedTopic2, 1, (short) 1);
    sharedKafkaTestResource
        .getKafkaTestUtils()
        .createTopic(expectedTopic3, 1, (short) 1);
    // Hit end point
    mockMvc
        .perform(get("/api/cluster/" + cluster.getId() + "/topics/list")
            .param("search", searchStr)
            .with(user(nonAdminUserDetails))
            .with(csrf())
            .contentType(MediaType.APPLICATION_JSON)
        )
        .andDo(print())
        .andExpect(status().isOk())
        // Validate Only the filtered instance shows up.
        .andExpect(content().string(containsString(expectedTopic2)))
        .andExpect(content().string(Matchers.not(containsString(expectedTopic1))))
        .andExpect(content().string(Matchers.not(containsString(expectedTopic3))));
}