private TransactionalTridentKafkaSpout createKafkaSpout() {

        ZkHosts hosts = new ZkHosts(zkUrl);

        TridentKafkaConfig config = new TridentKafkaConfig(hosts, "test");

        config.scheme = new SchemeAsMultiScheme(new StringScheme());



        config.startOffsetTime = kafka.api.OffsetRequest.LatestTime();

        return new TransactionalTridentKafkaSpout(config);

    }