conductor/kangaroo

Hadoop utilities for Kafka, S3, and more

kangaroo's Issues

Missing offsets on Leader re-election

I have found that Kangaroo is missing offsets and starts reading partition from the beginning when Leader for a partition gets changed.
That happens, because offsets are being stored under /kangaroo-consumers/<consumer_group>/<offsets or offsets-temp>/<topic>/<brokerId>-<partitionId>.
Since Leader re-election is quite usual situation for Kafka, I propose to remove brokerId in getBrokerPartition() from com.conductor.kafka.Partition:
public String getBrokerPartition() {
return String.format("%d-%d", broker.getId(), partId);
}

Kafka 0.8 Zookeeper DataStructures

Kafka 0.8 has new Zookeeper Data Structures and as a result ZKUtils is not working with 0.8+. See: https://cwiki.apache.org/confluence/display/KAFKA/Kafka+data+structures+in+Zookeeper

Using zookeeper for offset management is deprecated. I'm wondering if there are any thoughts on this or maybe even work already done. If not, I'm willing to give this a go.

Maven repository

Hello,

I would like to use kangaroo in one of my projects.
I was wondering if kangaroo is currently published on a maven repository, is it? If not, is it planned to be?

Thanks!

Kangaroo for kafka 0.8 doesnt work

Hi, I tried to implement the kangaroo in kafka 0.8.2.2, it seems not working.
When I tried to debug it, it cannot find the partitions. But I am sure, I have set it up correctly.

Configuration conf = getConf();

        //initializeConfig(conf);
        // Create a new job
        final Job job = Job.getInstance(conf, "TestMr");
        job.setJarByClass(App.class);

// Set the InputFormat
job.setInputFormatClass(KafkaInputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(BytesWritable.class);
job.setMapperClass(KafkaMapper.class);
job.setReducerClass(DnaReducer.class);
FileSystem hdfs = FileSystem.get(conf);
job.setNumReduceTasks(1000);
if (hdfs.exists(new Path("out")))
hdfs.delete(new Path("out"), true);
// Set your Zookeeper connection string
KafkaInputFormat.setZkConnect(job, "place_holder_for_kafka_server:2181");

// Set the topic you want to consume
KafkaInputFormat.setZkRoot(job, "/");
KafkaInputFormat.setTopic(job, "topic");
//KafkaInputFormat.setMaxSplitsPerPartition(job, 5);

// Set the consumer group associated with this job
KafkaInputFormat.setConsumerGroup(job, "mygroup");
FileOutputFormat.setOutputPath(job, new Path("out"));
// Set the mapper that will consume the data

// (Optional) Only commit offsets if the job is successful
if (job.waitForCompletion(true)) {
final ZkUtils zk = new ZkUtils(job.getConfiguration());
zk.commit("mygroup", "topic");
zk.close();
}
}
catch (Exception ex)
{
ex.printStackTrace();
}

when I implement the MR, there is no error, and it seems cannot find any partition from kafka.