NIFI-1233 upgraded to Kafka 0.9.0.0

Signed-off-by: jpercivall <joepercivall@yahoo.com>
This commit is contained in:
Oleg Zhurakousky 2016-01-12 10:56:49 -05:00 committed by jpercivall
parent 287d3ef531
commit 133838a93f
4 changed files with 19 additions and 7 deletions

View File

@ -37,12 +37,12 @@
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.8.2.2</version>
<version>0.9.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.9.1</artifactId>
<version>0.8.2.2</version>
<artifactId>kafka_2.10</artifactId>
<version>0.9.0.0</version >
<exclusions>
<!-- Transitive dependencies excluded because they are located
in a legacy Maven repository, which Maven 3 doesn't support. -->

View File

@ -25,6 +25,7 @@ import org.I0Itec.zkclient.serialize.ZkSerializer;
import kafka.admin.AdminUtils;
import kafka.api.TopicMetadata;
import kafka.utils.ZKStringSerializer;
import kafka.utils.ZkUtils;
import scala.collection.JavaConversions;
/**
@ -38,6 +39,7 @@ class KafkaUtils {
*/
static int retrievePartitionCountForTopic(String zookeeperConnectionString, String topicName) {
ZkClient zkClient = new ZkClient(zookeeperConnectionString);
zkClient.setZkSerializer(new ZkSerializer() {
@Override
public byte[] serialize(Object o) throws ZkMarshallingError {
@ -50,7 +52,7 @@ class KafkaUtils {
}
});
scala.collection.Set<TopicMetadata> topicMetadatas = AdminUtils
.fetchTopicMetadataFromZk(JavaConversions.asScalaSet(Collections.singleton(topicName)), zkClient);
.fetchTopicMetadataFromZk(JavaConversions.asScalaSet(Collections.singleton(topicName)), ZkUtils.apply(zkClient, false));
return topicMetadatas.size();
}
}

View File

@ -20,14 +20,15 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@ -75,8 +76,6 @@ import org.apache.nifi.stream.io.StreamUtils;
import org.apache.nifi.stream.io.util.NonThreadSafeCircularBuffer;
import org.apache.nifi.util.LongHolder;
import scala.actors.threadpool.Arrays;
@SupportsBatching
@InputRequirement(Requirement.INPUT_REQUIRED)
@Tags({"Apache", "Kafka", "Put", "Send", "Message", "PubSub"})

View File

@ -27,6 +27,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.producer.BufferExhaustedException;
import org.apache.kafka.clients.producer.Callback;
@ -474,6 +475,16 @@ public class TestPutKafka {
@Override
public void close() {
}
@Override
public void close(long arg0, TimeUnit arg1) {
// ignore, not used in test
}
@Override
public void flush() {
// ignore, not used in test
}
}
}