@Test
public void testMigrateZookeeperOffsetsWhenTopicNotExists() throws Exception {

  String topic = findUnusedTopic();
  Context context = prepareDefaultContext("testMigrateOffsets-nonExistingTopic");
  context.put(ZOOKEEPER_CONNECT_FLUME_KEY, kafkaServer.getZkConnectString());
  context.put(TOPIC, topic);
  KafkaSource source = new KafkaSource();
  source.doConfigure(context);
  source.setChannelProcessor(createGoodChannel());
  source.start();
  assertEquals(LifecycleState.START, source.getLifecycleState());
  Status status = source.process();
  assertEquals(Status.BACKOFF, status);
  source.stop();
}