  @Test
  public void createTopic_nonExistingCluster_throwsNotFound() throws Exception {

    expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.empty()));
    replay(clusterManager);

    try {
      topicManager.createTopic(
          CLUSTER_ID,
          TOPIC_1.getName(),
          TOPIC_1.getPartitions().size(),
          TOPIC_1.getReplicationFactor(),
          singletonMap("cleanup.policy", "compact")).get();
      fail();
    } catch (ExecutionException e) {
      assertEquals(NotFoundException.class, e.getCause().getClass());
    }
  }