You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@twill.apache.org by ch...@apache.org on 2014/04/22 08:06:02 UTC
[05/50] [abbrv] [TWILL-27] Upgrade to Kafka-0.8
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaClient.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaClient.java b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaClient.java
deleted file mode 100644
index 8ff4856..0000000
--- a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaClient.java
+++ /dev/null
@@ -1,304 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.twill.internal.kafka.client;
-
-import org.apache.twill.common.Threads;
-import org.apache.twill.kafka.client.FetchException;
-import org.apache.twill.kafka.client.FetchedMessage;
-import org.apache.twill.kafka.client.KafkaClient;
-import org.apache.twill.kafka.client.PreparePublish;
-import org.apache.twill.zookeeper.ZKClient;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.primitives.Ints;
-import com.google.common.primitives.Longs;
-import com.google.common.util.concurrent.AbstractIdleService;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
-import org.jboss.netty.bootstrap.ClientBootstrap;
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.buffer.ChannelBuffers;
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelFuture;
-import org.jboss.netty.channel.ChannelFutureListener;
-import org.jboss.netty.channel.ChannelPipeline;
-import org.jboss.netty.channel.ChannelPipelineFactory;
-import org.jboss.netty.channel.Channels;
-import org.jboss.netty.channel.socket.nio.NioClientBossPool;
-import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
-import org.jboss.netty.channel.socket.nio.NioWorkerPool;
-import org.jboss.netty.util.HashedWheelTimer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.nio.ByteBuffer;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-
-/**
- * Basic implementation of {@link KafkaClient}.
- */
-public final class SimpleKafkaClient extends AbstractIdleService implements KafkaClient {
-
- private static final Logger LOG = LoggerFactory.getLogger(SimpleKafkaClient.class);
- private static final int BROKER_POLL_INTERVAL = 100;
-
- private final KafkaBrokerCache brokerCache;
- private ClientBootstrap bootstrap;
- private ConnectionPool connectionPool;
-
- public SimpleKafkaClient(ZKClient zkClient) {
- this.brokerCache = new KafkaBrokerCache(zkClient);
- }
-
- @Override
- protected void startUp() throws Exception {
- brokerCache.startAndWait();
- ThreadFactory threadFactory = Threads.createDaemonThreadFactory("kafka-client-netty-%d");
- NioClientBossPool bossPool = new NioClientBossPool(Executors.newSingleThreadExecutor(threadFactory), 1,
- new HashedWheelTimer(threadFactory), null);
- NioWorkerPool workerPool = new NioWorkerPool(Executors.newFixedThreadPool(4, threadFactory), 4);
-
- bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(bossPool, workerPool));
- bootstrap.setPipelineFactory(new KafkaChannelPipelineFactory());
- connectionPool = new ConnectionPool(bootstrap);
- }
-
- @Override
- protected void shutDown() throws Exception {
- connectionPool.close();
- bootstrap.releaseExternalResources();
- brokerCache.stopAndWait();
- }
-
- @Override
- public PreparePublish preparePublish(final String topic, final Compression compression) {
- final Map<Integer, MessageSetEncoder> encoders = Maps.newHashMap();
-
- return new PreparePublish() {
- @Override
- public PreparePublish add(byte[] payload, Object partitionKey) {
- return add(ByteBuffer.wrap(payload), partitionKey);
- }
-
- @Override
- public PreparePublish add(ByteBuffer payload, Object partitionKey) {
- // TODO: Partition
- int partition = 0;
-
- MessageSetEncoder encoder = encoders.get(partition);
- if (encoder == null) {
- encoder = getEncoder(compression);
- encoders.put(partition, encoder);
- }
- encoder.add(ChannelBuffers.wrappedBuffer(payload));
-
- return this;
- }
-
- @Override
- public ListenableFuture<?> publish() {
- List<ListenableFuture<?>> futures = Lists.newArrayListWithCapacity(encoders.size());
- for (Map.Entry<Integer, MessageSetEncoder> entry : encoders.entrySet()) {
- futures.add(doPublish(topic, entry.getKey(), entry.getValue().finish()));
- }
- encoders.clear();
- return Futures.allAsList(futures);
- }
-
- private ListenableFuture<?> doPublish(String topic, int partition, ChannelBuffer messageSet) {
- final KafkaRequest request = KafkaRequest.createProduce(topic, partition, messageSet);
- final SettableFuture<?> result = SettableFuture.create();
- final ConnectionPool.ConnectResult connection =
- connectionPool.connect(getTopicBroker(topic, partition).getAddress());
-
- connection.getChannelFuture().addListener(new ChannelFutureListener() {
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- try {
- future.getChannel().write(request).addListener(getPublishChannelFutureListener(result, null, connection));
- } catch (Exception e) {
- result.setException(e);
- }
- }
- });
-
- return result;
- }
- };
- }
-
- @Override
- public Iterator<FetchedMessage> consume(final String topic, final int partition, long offset, int maxSize) {
- Preconditions.checkArgument(maxSize >= 10, "Message size cannot be smaller than 10.");
-
- // Connect to broker. Consumer connection are long connection. No need to worry about reuse.
- final AtomicReference<ChannelFuture> channelFutureRef = new AtomicReference<ChannelFuture>(
- connectionPool.connect(getTopicBroker(topic, partition).getAddress()).getChannelFuture());
-
- return new MessageFetcher(topic, partition, offset, maxSize, new KafkaRequestSender() {
-
- @Override
- public void send(final KafkaRequest request) {
- if (!isRunning()) {
- return;
- }
- try {
- // Try to send the request
- Channel channel = channelFutureRef.get().getChannel();
- if (!channel.write(request).await().isSuccess()) {
- // If failed, retry
- channel.close();
- ChannelFuture channelFuture = connectionPool.connect(
- getTopicBroker(topic, partition).getAddress()).getChannelFuture();
- channelFutureRef.set(channelFuture);
- channelFuture.addListener(new ChannelFutureListener() {
- @Override
- public void operationComplete(ChannelFuture channelFuture) throws Exception {
- send(request);
- }
- });
- }
- } catch (InterruptedException e) {
- // Ignore it
- LOG.info("Interrupted when sending consume request", e);
- }
- }
- });
- }
-
- @Override
- public ListenableFuture<long[]> getOffset(final String topic, final int partition, long time, int maxOffsets) {
- final SettableFuture<long[]> resultFuture = SettableFuture.create();
- final ChannelBuffer body = ChannelBuffers.buffer(Longs.BYTES + Ints.BYTES);
- body.writeLong(time);
- body.writeInt(maxOffsets);
-
- connectionPool.connect(getTopicBroker(topic, partition).getAddress())
- .getChannelFuture().addListener(new ChannelFutureListener() {
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- if (checkFailure(future)) {
- return;
- }
-
- future.getChannel().write(KafkaRequest.createOffsets(topic, partition, body, new ResponseHandler() {
- @Override
- public void received(KafkaResponse response) {
- if (response.getErrorCode() != FetchException.ErrorCode.OK) {
- resultFuture.setException(new FetchException("Failed to fetch offset.", response.getErrorCode()));
- } else {
- // Decode the offset response, which contains 4 bytes number of offsets, followed by number of offsets,
- // each 8 bytes in size.
- ChannelBuffer resultBuffer = response.getBody();
- int size = resultBuffer.readInt();
- long[] result = new long[size];
- for (int i = 0; i < size; i++) {
- result[i] = resultBuffer.readLong();
- }
- resultFuture.set(result);
- }
- }
- })).addListener(new ChannelFutureListener() {
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- checkFailure(future);
- }
- });
- }
-
- private boolean checkFailure(ChannelFuture future) {
- if (!future.isSuccess()) {
- if (future.isCancelled()) {
- resultFuture.cancel(true);
- } else {
- resultFuture.setException(future.getCause());
- }
- return true;
- }
- return false;
- }
- });
-
- return resultFuture;
- }
-
- private TopicBroker getTopicBroker(String topic, int partition) {
- TopicBroker topicBroker = brokerCache.getBrokerAddress(topic, partition);
- while (topicBroker == null) {
- try {
- TimeUnit.MILLISECONDS.sleep(BROKER_POLL_INTERVAL);
- } catch (InterruptedException e) {
- return null;
- }
- topicBroker = brokerCache.getBrokerAddress(topic, partition);
- }
- return topicBroker;
- }
-
- private MessageSetEncoder getEncoder(Compression compression) {
- switch (compression) {
- case GZIP:
- return new GZipMessageSetEncoder();
- case SNAPPY:
- return new SnappyMessageSetEncoder();
- default:
- return new IdentityMessageSetEncoder();
- }
- }
-
- private <V> ChannelFutureListener getPublishChannelFutureListener(final SettableFuture<V> result, final V resultObj,
- final ConnectionPool.ConnectionReleaser releaser) {
- return new ChannelFutureListener() {
- @Override
- public void operationComplete(ChannelFuture future) throws Exception {
- try {
- if (future.isSuccess()) {
- result.set(resultObj);
- } else if (future.isCancelled()) {
- result.cancel(true);
- } else {
- result.setException(future.getCause());
- }
- } finally {
- releaser.release();
- }
- }
- };
- }
-
- private static final class KafkaChannelPipelineFactory implements ChannelPipelineFactory {
-
- @Override
- public ChannelPipeline getPipeline() throws Exception {
- ChannelPipeline pipeline = Channels.pipeline();
-
- pipeline.addLast("encoder", new KafkaRequestEncoder());
- pipeline.addLast("decoder", new KafkaResponseHandler());
- pipeline.addLast("dispatcher", new KafkaResponseDispatcher());
- return pipeline;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaConsumer.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaConsumer.java b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaConsumer.java
new file mode 100644
index 0000000..230521c
--- /dev/null
+++ b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaConsumer.java
@@ -0,0 +1,481 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.internal.kafka.client;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
+import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.Futures;
+import kafka.api.FetchRequest;
+import kafka.api.FetchRequestBuilder;
+import kafka.api.PartitionOffsetRequestInfo;
+import kafka.common.ErrorMapping;
+import kafka.common.TopicAndPartition;
+import kafka.javaapi.FetchResponse;
+import kafka.javaapi.OffsetRequest;
+import kafka.javaapi.OffsetResponse;
+import kafka.javaapi.consumer.SimpleConsumer;
+import kafka.javaapi.message.ByteBufferMessageSet;
+import kafka.message.MessageAndOffset;
+import org.apache.twill.common.Cancellable;
+import org.apache.twill.common.Threads;
+import org.apache.twill.kafka.client.BrokerInfo;
+import org.apache.twill.kafka.client.BrokerService;
+import org.apache.twill.kafka.client.FetchedMessage;
+import org.apache.twill.kafka.client.KafkaConsumer;
+import org.apache.twill.kafka.client.TopicPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.channels.ClosedByInterruptException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * A {@link KafkaConsumer} implementation using the scala kafka api.
+ */
+final class SimpleKafkaConsumer implements KafkaConsumer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SimpleKafkaConsumer.class);
+ private static final int FETCH_SIZE = 1024 * 1024; // Use a default fetch size.
+ private static final int SO_TIMEOUT = 5 * 1000; // 5 seconds.
+ private static final int MAX_WAIT = 1000; // 1 second.
+ private static final long CONSUMER_EXPIRE_MINUTES = 1L; // close consumer if not used for 1 minute.
+ private static final long CONSUMER_FAILURE_RETRY_INTERVAL = 2000L; // Sleep for 2 seconds if failure in consumer.
+ private static final long EMPTY_FETCH_WAIT = 500L; // Sleep for 500 ms if no message is fetched.
+
+ private final BrokerService brokerService;
+ private final LoadingCache<BrokerInfo, SimpleConsumer> consumers;
+ private final BlockingQueue<Cancellable> consumerCancels;
+
+ SimpleKafkaConsumer(BrokerService brokerService) {
+ this.brokerService = brokerService;
+ this.consumers = CacheBuilder.newBuilder()
+ .expireAfterAccess(CONSUMER_EXPIRE_MINUTES, TimeUnit.MINUTES)
+ .removalListener(createRemovalListener())
+ .build(createConsumerLoader());
+ this.consumerCancels = new LinkedBlockingQueue<Cancellable>();
+ }
+
+ @Override
+ public Preparer prepare() {
+ return new SimplePreparer();
+ }
+
+ /**
+ * Called to stop all consumers created. This method should only be
+ * called by KafkaClientService who own this consumer.
+ */
+ void stop() {
+ LOG.info("Stopping Kafka consumer");
+ List<Cancellable> cancels = Lists.newLinkedList();
+ consumerCancels.drainTo(cancels);
+ for (Cancellable cancel : cancels) {
+ cancel.cancel();
+ }
+ consumers.invalidateAll();
+ LOG.info("Kafka Consumer stopped");
+ }
+
+ /**
+ * Creates a CacheLoader for creating SimpleConsumer.
+ */
+ private CacheLoader<BrokerInfo, SimpleConsumer> createConsumerLoader() {
+ return new CacheLoader<BrokerInfo, SimpleConsumer>() {
+ @Override
+ public SimpleConsumer load(BrokerInfo key) throws Exception {
+ return new SimpleConsumer(key.getHost(), key.getPort(), SO_TIMEOUT, FETCH_SIZE, "simple-kafka-client");
+ }
+ };
+ }
+
+ /**
+ * Creates a RemovalListener that will close SimpleConsumer on cache removal.
+ */
+ private RemovalListener<BrokerInfo, SimpleConsumer> createRemovalListener() {
+ return new RemovalListener<BrokerInfo, SimpleConsumer>() {
+ @Override
+ public void onRemoval(RemovalNotification<BrokerInfo, SimpleConsumer> notification) {
+ SimpleConsumer consumer = notification.getValue();
+ if (consumer != null) {
+ consumer.close();
+ }
+ }
+ };
+ }
+
+ /**
+ * Retrieves the last offset before the given timestamp for a given topic partition.
+ *
+ * @return The last offset before the given timestamp or {@code 0} if failed to do so.
+ */
+ private long getLastOffset(TopicPartition topicPart, long timestamp) {
+ BrokerInfo brokerInfo = brokerService.getLeader(topicPart.getTopic(), topicPart.getPartition());
+ SimpleConsumer consumer = brokerInfo == null ? null : consumers.getUnchecked(brokerInfo);
+
+ // If no broker, treat it as failure attempt.
+ if (consumer == null) {
+ LOG.warn("Failed to talk to any broker. Default offset to 0 for {}", topicPart);
+ return 0L;
+ }
+
+ // Fire offset request
+ OffsetRequest request = new OffsetRequest(ImmutableMap.of(
+ new TopicAndPartition(topicPart.getTopic(), topicPart.getPartition()),
+ new PartitionOffsetRequestInfo(timestamp, 1)
+ ), kafka.api.OffsetRequest.CurrentVersion(), consumer.clientId());
+
+ OffsetResponse response = consumer.getOffsetsBefore(request);
+
+ // Retrieve offsets from response
+ long[] offsets = response.hasError() ? null : response.offsets(topicPart.getTopic(), topicPart.getPartition());
+ if (offsets == null || offsets.length <= 0) {
+ short errorCode = response.errorCode(topicPart.getTopic(), topicPart.getPartition());
+
+ // If the topic partition doesn't exists, use offset 0 without logging error.
+ if (errorCode != ErrorMapping.UnknownTopicOrPartitionCode()) {
+ consumers.refresh(brokerInfo);
+ LOG.warn("Failed to fetch offset for {} with timestamp {}. Error: {}. Default offset to 0.",
+ topicPart, timestamp, errorCode);
+ }
+ return 0L;
+ }
+
+ LOG.debug("Offset {} fetched for {} with timestamp {}.", offsets[0], topicPart, timestamp);
+ return offsets[0];
+ }
+
+ /**
+ * A preparer that uses kafak scala api for consuming messages.
+ */
+ private final class SimplePreparer implements Preparer {
+
+ // Map from TopicPartition to offset
+ private final Map<TopicPartition, Long> requests;
+ private final ThreadFactory threadFactory;
+
+ private SimplePreparer() {
+ this.requests = Maps.newHashMap();
+ this.threadFactory = Threads.createDaemonThreadFactory("message-callback-%d");
+ }
+
+ @Override
+ public Preparer add(String topic, int partition, long offset) {
+ requests.put(new TopicPartition(topic, partition), offset);
+ return this;
+ }
+
+ @Override
+ public Preparer addFromBeginning(String topic, int partition) {
+ TopicPartition topicPartition = new TopicPartition(topic, partition);
+ requests.put(topicPartition, kafka.api.OffsetRequest.EarliestTime());
+ return this;
+ }
+
+ @Override
+ public Preparer addLatest(String topic, int partition) {
+ TopicPartition topicPartition = new TopicPartition(topic, partition);
+ requests.put(topicPartition, kafka.api.OffsetRequest.LatestTime());
+ return this;
+ }
+
+ @Override
+ public Cancellable consume(MessageCallback callback) {
+ final ExecutorService executor = Executors.newSingleThreadExecutor(threadFactory);
+ final List<ConsumerThread> pollers = Lists.newArrayList();
+
+ // When cancelling the consumption, first terminates all polling threads and then stop the executor service.
+ final AtomicBoolean cancelled = new AtomicBoolean();
+ Cancellable cancellable = new Cancellable() {
+ @Override
+ public void cancel() {
+ if (!cancelled.compareAndSet(false, true)) {
+ return;
+ }
+ consumerCancels.remove(this);
+
+ LOG.info("Requesting stop of all consumer threads.");
+ for (ConsumerThread consumerThread : pollers) {
+ consumerThread.terminate();
+ }
+ LOG.info("Wait for all consumer threads to stop.");
+ for (ConsumerThread consumerThread : pollers) {
+ try {
+ consumerThread.join();
+ } catch (InterruptedException e) {
+ LOG.warn("Interrupted exception while waiting for thread to complete.", e);
+ }
+ }
+ LOG.info("All consumer threads stopped.");
+ // Use shutdown so that submitted task still has chance to execute, which is important for finished to get
+ // called.
+ executor.shutdown();
+ }
+ };
+
+ // Wrap the callback with a single thread executor.
+ MessageCallback messageCallback = wrapCallback(callback, executor, cancellable);
+
+ // Starts threads for polling new messages.
+ for (Map.Entry<TopicPartition, Long> entry : requests.entrySet()) {
+ ConsumerThread consumerThread = new ConsumerThread(entry.getKey(), entry.getValue(), messageCallback);
+ consumerThread.setDaemon(true);
+ consumerThread.start();
+ pollers.add(consumerThread);
+ }
+
+ consumerCancels.add(cancellable);
+ return cancellable;
+ }
+
+ /**
+ * Wrap a given MessageCallback by a executor so that calls are executed in the given executor.
+ * By running the calls through the executor, it also block and wait for the task being completed so that
+ * it can block the poller thread depending on the rate of processing that the callback can handle.
+ */
+ private MessageCallback wrapCallback(final MessageCallback callback,
+ final ExecutorService executor, final Cancellable cancellable) {
+ final AtomicBoolean stopped = new AtomicBoolean();
+ return new MessageCallback() {
+ @Override
+ public void onReceived(final Iterator<FetchedMessage> messages) {
+ if (stopped.get()) {
+ return;
+ }
+ Futures.getUnchecked(executor.submit(new Runnable() {
+ @Override
+ public void run() {
+ if (stopped.get()) {
+ return;
+ }
+ callback.onReceived(messages);
+ }
+ }));
+ }
+
+ @Override
+ public void finished() {
+ // Make sure finished only get called once.
+ if (!stopped.compareAndSet(false, true)) {
+ return;
+ }
+ Futures.getUnchecked(executor.submit(new Runnable() {
+ @Override
+ public void run() {
+ // When finished is called, also cancel the consumption from all polling thread.
+ callback.finished();
+ cancellable.cancel();
+ }
+ }));
+ }
+ };
+ }
+ }
+
+ /**
+ * The thread for polling kafka.
+ */
+ private final class ConsumerThread extends Thread {
+
+ private final TopicPartition topicPart;
+ private final long startOffset;
+ private final MessageCallback callback;
+ private final BasicFetchedMessage fetchedMessage;
+ private volatile boolean running;
+
+ private ConsumerThread(TopicPartition topicPart, long startOffset, MessageCallback callback) {
+ super(String.format("Kafka-Consumer-%s-%d", topicPart.getTopic(), topicPart.getPartition()));
+ this.topicPart = topicPart;
+ this.startOffset = startOffset;
+ this.callback = callback;
+ this.running = true;
+ this.fetchedMessage = new BasicFetchedMessage(topicPart);
+ }
+
+ @Override
+ public void run() {
+ final AtomicLong offset = new AtomicLong(startOffset);
+
+ Map.Entry<BrokerInfo, SimpleConsumer> consumerEntry = null;
+
+ while (running) {
+ if (consumerEntry == null && (consumerEntry = getConsumerEntry()) == null) {
+ LOG.debug("No leader for topic partition {}.", topicPart);
+ try {
+ TimeUnit.MILLISECONDS.sleep(CONSUMER_FAILURE_RETRY_INTERVAL);
+ } catch (InterruptedException e) {
+ // OK to ignore this, as interrupt would be caused by thread termination.
+ LOG.trace("Consumer sleep interrupted.", e);
+ }
+ continue;
+ }
+
+ // If offset < 0, meaning it's special offset value that needs to fetch either the earliest or latest offset
+ // from kafak server.
+ long off = offset.get();
+ if (off < 0) {
+ offset.set(getLastOffset(topicPart, off));
+ }
+
+ SimpleConsumer consumer = consumerEntry.getValue();
+
+ // Fire a fetch message request
+ try {
+ FetchResponse response = fetchMessages(consumer, offset.get());
+
+ // Failure response, set consumer entry to null and let next round of loop to handle it.
+ if (response.hasError()) {
+ short errorCode = response.errorCode(topicPart.getTopic(), topicPart.getPartition());
+ LOG.info("Failed to fetch message on {}. Error: {}", topicPart, errorCode);
+ // If it is out of range error, reset to earliest offset
+ if (errorCode == ErrorMapping.OffsetOutOfRangeCode()) {
+ offset.set(kafka.api.OffsetRequest.EarliestTime());
+ }
+
+ consumers.refresh(consumerEntry.getKey());
+ consumerEntry = null;
+ continue;
+ }
+
+ ByteBufferMessageSet messages = response.messageSet(topicPart.getTopic(), topicPart.getPartition());
+ if (sleepIfEmpty(messages)) {
+ continue;
+ }
+
+ // Call the callback
+ invokeCallback(messages, offset);
+ } catch (Throwable t) {
+ if (running || !(t instanceof ClosedByInterruptException)) {
+ // Only log if it is still running, otherwise, it just the interrupt caused by the stop.
+ LOG.info("Exception when fetching message on {}.", topicPart, t);
+ }
+ consumers.refresh(consumerEntry.getKey());
+ consumerEntry = null;
+ }
+ }
+
+ // When the thread is done, call the callback finished method.
+ try {
+ callback.finished();
+ } catch (Throwable t) {
+ LOG.error("Exception thrown from MessageCallback.finished({})", running, t);
+ }
+ }
+
+ public void terminate() {
+ LOG.info("Terminate requested {}", getName());
+ running = false;
+ interrupt();
+ }
+
+ /**
+ * Gets the leader broker and the associated SimpleConsumer for the current topic and partition.
+ */
+ private Map.Entry<BrokerInfo, SimpleConsumer> getConsumerEntry() {
+ BrokerInfo leader = brokerService.getLeader(topicPart.getTopic(), topicPart.getPartition());
+ return leader == null ? null : Maps.immutableEntry(leader, consumers.getUnchecked(leader));
+ }
+
+ /**
+ * Makes a call to kafka to fetch messages.
+ */
+ private FetchResponse fetchMessages(SimpleConsumer consumer, long offset) {
+ FetchRequest request = new FetchRequestBuilder()
+ .clientId(consumer.clientId())
+ .addFetch(topicPart.getTopic(), topicPart.getPartition(), offset, FETCH_SIZE)
+ .maxWait(MAX_WAIT)
+ .build();
+ return consumer.fetch(request);
+ }
+
+ /**
+ * Sleeps if the message set is empty.
+ * @return {@code true} if it is empty, {@code false} otherwise.
+ */
+ private boolean sleepIfEmpty(ByteBufferMessageSet messages) {
+ if (Iterables.isEmpty(messages)) {
+ LOG.trace("No message fetched. Sleep for {} ms before next fetch.", EMPTY_FETCH_WAIT);
+ try {
+ TimeUnit.MILLISECONDS.sleep(EMPTY_FETCH_WAIT);
+ } catch (InterruptedException e) {
+ // It's interrupted from stop, ok to ignore.
+ }
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Calls the message callback with the given message set.
+ */
+ private void invokeCallback(ByteBufferMessageSet messages, AtomicLong offset) {
+ long savedOffset = offset.get();
+ try {
+ callback.onReceived(createFetchedMessages(messages, offset));
+ } catch (Throwable t) {
+ LOG.error("Callback throws exception. Retry from offset {} for {}", startOffset, topicPart, t);
+ offset.set(savedOffset);
+ }
+ }
+
+ /**
+ * Creates an Iterator of FetchedMessage based on the given message set. The iterator would also updates
+ * the offset while iterating.
+ */
+ private Iterator<FetchedMessage> createFetchedMessages(ByteBufferMessageSet messageSet, final AtomicLong offset) {
+ final Iterator<MessageAndOffset> messages = messageSet.iterator();
+ return new AbstractIterator<FetchedMessage>() {
+ @Override
+ protected FetchedMessage computeNext() {
+ while (messages.hasNext()) {
+ MessageAndOffset message = messages.next();
+ long msgOffset = message.offset();
+ if (msgOffset < offset.get()) {
+ LOG.trace("Received old offset {}, expecting {} on {}. Message Ignored.",
+ msgOffset, offset.get(), topicPart);
+ continue;
+ }
+
+ offset.set(message.nextOffset());
+ fetchedMessage.setPayload(message.message().payload());
+ fetchedMessage.setNextOffset(offset.get());
+
+ return fetchedMessage;
+ }
+ return endOfData();
+ }
+ };
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaPublisher.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaPublisher.java b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaPublisher.java
new file mode 100644
index 0000000..97b14c6
--- /dev/null
+++ b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SimpleKafkaPublisher.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.internal.kafka.client;
+
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import kafka.javaapi.producer.Producer;
+import kafka.producer.KeyedMessage;
+import kafka.producer.ProducerConfig;
+import org.apache.twill.common.Cancellable;
+import org.apache.twill.kafka.client.Compression;
+import org.apache.twill.kafka.client.KafkaPublisher;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Properties;
+
+/**
+ * Implementation of {@link KafkaPublisher} using the kafka scala-java api.
+ */
+final class SimpleKafkaPublisher implements KafkaPublisher {
+
+ private final String kafkaBrokers;
+ private final Ack ack;
+ private final Compression compression;
+ private Producer<Integer, ByteBuffer> producer;
+
+ public SimpleKafkaPublisher(String kafkaBrokers, Ack ack, Compression compression) {
+ this.kafkaBrokers = kafkaBrokers;
+ this.ack = ack;
+ this.compression = compression;
+ }
+
+ /**
+ * Start the publisher. This method must be called before other methods. This method is only to be called
+ * by KafkaClientService who own this object.
+ * @return A Cancellable for closing this publish.
+ */
+ Cancellable start() {
+ // It should return a Cancellable that is not holding any reference to this instance.
+ // It is for ZKKafkaClientService to close the internal Kafka producer when this publisher get garbage collected.
+ Properties props = new Properties();
+ props.put("metadata.broker.list", kafkaBrokers);
+ props.put("serializer.class", ByteBufferEncoder.class.getName());
+ props.put("key.serializer.class", IntegerEncoder.class.getName());
+ props.put("partitioner.class", IntegerPartitioner.class.getName());
+ props.put("request.required.acks", Integer.toString(ack.getAck()));
+ props.put("compression.codec", compression.getCodec());
+
+ producer = new Producer<Integer, ByteBuffer>(new ProducerConfig(props));
+ return new ProducerCancellable(producer);
+ }
+
+ @Override
+ public Preparer prepare(String topic) {
+ return new SimplePreparer(topic);
+ }
+
+ private final class SimplePreparer implements Preparer {
+
+ private final String topic;
+ private final List<KeyedMessage<Integer, ByteBuffer>> messages;
+
+ private SimplePreparer(String topic) {
+ this.topic = topic;
+ this.messages = Lists.newLinkedList();
+ }
+
+ @Override
+ public Preparer add(ByteBuffer message, Object partitionKey) {
+ messages.add(new KeyedMessage<Integer, ByteBuffer>(topic, Math.abs(partitionKey.hashCode()), message));
+ return this;
+ }
+
+ @Override
+ public ListenableFuture<Integer> send() {
+ int size = messages.size();
+ producer.send(messages);
+
+ messages.clear();
+ return Futures.immediateFuture(size);
+ }
+ }
+
+ private static final class ProducerCancellable implements Cancellable {
+ private final Producer<Integer, ByteBuffer> producer;
+
+ private ProducerCancellable(Producer<Integer, ByteBuffer> producer) {
+ this.producer = producer;
+ }
+
+ @Override
+ public void cancel() {
+ producer.close();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SnappyMessageSetEncoder.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SnappyMessageSetEncoder.java b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SnappyMessageSetEncoder.java
deleted file mode 100644
index bf18c08..0000000
--- a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/SnappyMessageSetEncoder.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.twill.internal.kafka.client;
-
-import org.xerial.snappy.SnappyOutputStream;
-
-import java.io.IOException;
-import java.io.OutputStream;
-
-/**
- * A {@link MessageSetEncoder} that compress messages using snappy.
- */
-final class SnappyMessageSetEncoder extends AbstractCompressedMessageSetEncoder {
-
- SnappyMessageSetEncoder() {
- super(Compression.SNAPPY);
- }
-
- @Override
- protected OutputStream createCompressedStream(OutputStream os) throws IOException {
- return new SnappyOutputStream(os);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/internal/kafka/client/TopicBroker.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/TopicBroker.java b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/TopicBroker.java
deleted file mode 100644
index fd4bf03..0000000
--- a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/TopicBroker.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.twill.internal.kafka.client;
-
-import java.net.InetSocketAddress;
-
-/**
- * Represents broker information of a given topic.
- */
-final class TopicBroker {
-
- private final String topic;
- private final InetSocketAddress address;
- private final int partitionSize;
-
- TopicBroker(String topic, InetSocketAddress address, int partitionSize) {
- this.topic = topic;
- this.address = address;
- this.partitionSize = partitionSize;
- }
-
- String getTopic() {
- return topic;
- }
-
- InetSocketAddress getAddress() {
- return address;
- }
-
- int getPartitionSize() {
- return partitionSize;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/internal/kafka/client/ZKBrokerService.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/ZKBrokerService.java b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/ZKBrokerService.java
new file mode 100644
index 0000000..74ecfc1
--- /dev/null
+++ b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/ZKBrokerService.java
@@ -0,0 +1,403 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.internal.kafka.client;
+
+import com.google.common.base.Charsets;
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
+import com.google.common.base.Throwables;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.primitives.Ints;
+import com.google.common.util.concurrent.AbstractIdleService;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.SettableFuture;
+import com.google.gson.Gson;
+import org.apache.twill.common.Threads;
+import org.apache.twill.kafka.client.BrokerInfo;
+import org.apache.twill.kafka.client.BrokerService;
+import org.apache.twill.kafka.client.TopicPartition;
+import org.apache.twill.zookeeper.NodeChildren;
+import org.apache.twill.zookeeper.NodeData;
+import org.apache.twill.zookeeper.ZKClient;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * A {@link BrokerService} that watches kafka zk nodes for updates of broker lists and leader for
+ * each topic partition.
+ */
+final class ZKBrokerService extends AbstractIdleService implements BrokerService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ZKBrokerService.class);
+ private static final String BROKER_IDS_PATH = "/brokers/ids";
+ private static final String BROKER_TOPICS_PATH = "/brokers/topics";
+ private static final long FAILURE_RETRY_SECONDS = 5;
+ private static final Gson GSON = new Gson();
+ private static final Function<String, BrokerId> BROKER_ID_TRANSFORMER = new Function<String, BrokerId>() {
+ @Override
+ public BrokerId apply(String input) {
+ return new BrokerId(Integer.parseInt(input));
+ }
+ };
+ private static final Function<BrokerInfo, String> BROKER_INFO_TO_ADDRESS = new Function<BrokerInfo, String>() {
+ @Override
+ public String apply(BrokerInfo input) {
+ return String.format("%s:%d", input.getHost(), input.getPort());
+ }
+ };
+
+ private final ZKClient zkClient;
+ private final LoadingCache<BrokerId, Supplier<BrokerInfo>> brokerInfos;
+ private final LoadingCache<KeyPathTopicPartition, Supplier<PartitionInfo>> partitionInfos;
+ // A comma separated list of brokers (host:port,host:port)
+
+ private ExecutorService executorService;
+ private Supplier<Iterable<BrokerInfo>> brokerList;
+
+ ZKBrokerService(ZKClient zkClient) {
+ this.zkClient = zkClient;
+ this.brokerInfos = CacheBuilder.newBuilder().build(createCacheLoader(new CacheInvalidater<BrokerId>() {
+ @Override
+ public void invalidate(BrokerId key) {
+ brokerInfos.invalidate(key);
+ }
+ }, BrokerInfo.class));
+ this.partitionInfos = CacheBuilder.newBuilder().build(createCacheLoader(
+ new CacheInvalidater<KeyPathTopicPartition>() {
+ @Override
+ public void invalidate(KeyPathTopicPartition key) {
+ partitionInfos.invalidate(key);
+ }
+ }, PartitionInfo.class));
+ }
+
+ @Override
+ protected void startUp() throws Exception {
+ executorService = Executors.newCachedThreadPool(Threads.createDaemonThreadFactory("zk-kafka-broker"));
+ }
+
+ @Override
+ protected void shutDown() throws Exception {
+ executorService.shutdownNow();
+ }
+
+ @Override
+ public BrokerInfo getLeader(String topic, int partition) {
+ Preconditions.checkState(isRunning(), "BrokerService is not running.");
+ PartitionInfo partitionInfo = partitionInfos.getUnchecked(new KeyPathTopicPartition(topic, partition)).get();
+ return partitionInfo == null ? null : brokerInfos.getUnchecked(new BrokerId(partitionInfo.getLeader())).get();
+ }
+
+ @Override
+ public synchronized Iterable<BrokerInfo> getBrokers() {
+ Preconditions.checkState(isRunning(), "BrokerService is not running.");
+
+ if (brokerList != null) {
+ return brokerList.get();
+ }
+
+ final SettableFuture<?> readerFuture = SettableFuture.create();
+ final AtomicReference<Iterable<BrokerInfo>> brokers =
+ new AtomicReference<Iterable<BrokerInfo>>(ImmutableList.<BrokerInfo>of());
+
+ actOnExists(BROKER_IDS_PATH, new Runnable() {
+ @Override
+ public void run() {
+ // Callback for fetching children list. This callback should be executed in the executorService.
+ final FutureCallback<NodeChildren> childrenCallback = new FutureCallback<NodeChildren>() {
+ @Override
+ public void onSuccess(NodeChildren result) {
+ try {
+ // For each children node, get the BrokerInfo from the brokerInfo cache.
+ brokers.set(
+ ImmutableList.copyOf(
+ Iterables.transform(
+ brokerInfos.getAll(Iterables.transform(result.getChildren(), BROKER_ID_TRANSFORMER)).values(),
+ Suppliers.<BrokerInfo>supplierFunction())));
+ readerFuture.set(null);
+
+ } catch (ExecutionException e) {
+ readerFuture.setException(e.getCause());
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ readerFuture.setException(t);
+ }
+ };
+
+ // Fetch list of broker ids
+ Futures.addCallback(zkClient.getChildren(BROKER_IDS_PATH, new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ if (!isRunning()) {
+ return;
+ }
+ if (event.getType() == Event.EventType.NodeChildrenChanged) {
+ Futures.addCallback(zkClient.getChildren(BROKER_IDS_PATH, this), childrenCallback, executorService);
+ }
+ }
+ }), childrenCallback, executorService);
+ }
+ }, readerFuture, FAILURE_RETRY_SECONDS, TimeUnit.SECONDS);
+
+ brokerList = createSupplier(brokers);
+ try {
+ readerFuture.get();
+ } catch (Exception e) {
+ throw Throwables.propagate(e);
+ }
+ return brokerList.get();
+ }
+
+ @Override
+ public String getBrokerList() {
+ return Joiner.on(',').join(Iterables.transform(getBrokers(), BROKER_INFO_TO_ADDRESS));
+ }
+
+ /**
+ * Creates a cache loader for the given path to supply data with the data node.
+ */
+ private <K extends KeyPath, T> CacheLoader<K, Supplier<T>> createCacheLoader(final CacheInvalidater<K> invalidater,
+ final Class<T> resultType) {
+ return new CacheLoader<K, Supplier<T>>() {
+
+ @Override
+ public Supplier<T> load(final K key) throws Exception {
+ // A future to tell if the result is ready, even it is failure.
+ final SettableFuture<T> readyFuture = SettableFuture.create();
+ final AtomicReference<T> resultValue = new AtomicReference<T>();
+
+ // Fetch for node data when it exists.
+ final String path = key.getPath();
+ actOnExists(path, new Runnable() {
+ @Override
+ public void run() {
+ // Callback for getData call
+ final FutureCallback<NodeData> dataCallback = new FutureCallback<NodeData>() {
+ @Override
+ public void onSuccess(NodeData result) {
+ // Update with latest data
+ T value = decodeNodeData(result, resultType);
+ resultValue.set(value);
+ readyFuture.set(value);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ LOG.error("Failed to fetch node data on {}", path, t);
+ if (t instanceof KeeperException.NoNodeException) {
+ resultValue.set(null);
+ readyFuture.set(null);
+ return;
+ }
+
+ // On error, simply invalidate the key so that it'll be fetched next time.
+ invalidater.invalidate(key);
+ readyFuture.setException(t);
+ }
+ };
+
+ // Fetch node data
+ Futures.addCallback(zkClient.getData(path, new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ if (!isRunning()) {
+ return;
+ }
+ if (event.getType() == Event.EventType.NodeDataChanged) {
+ // If node data changed, fetch it again.
+ Futures.addCallback(zkClient.getData(path, this), dataCallback, executorService);
+ } else if (event.getType() == Event.EventType.NodeDeleted) {
+ // If node removed, invalidate the cached value.
+ brokerInfos.invalidate(key);
+ }
+ }
+ }), dataCallback, executorService);
+ }
+ }, readyFuture, FAILURE_RETRY_SECONDS, TimeUnit.SECONDS);
+
+ readyFuture.get();
+ return createSupplier(resultValue);
+ }
+ };
+ }
+
+ /**
+ * Gson decode the NodeData into object.
+ * @param nodeData The data to decode
+ * @param type Object class to decode into.
+ * @param <T> Type of the object.
+ * @return The decoded object or {@code null} if node data is null.
+ */
+ private <T> T decodeNodeData(NodeData nodeData, Class<T> type) {
+ byte[] data = nodeData == null ? null : nodeData.getData();
+ if (data == null) {
+ return null;
+ }
+ return GSON.fromJson(new String(data, Charsets.UTF_8), type);
+ }
+
+ /**
+ * Checks exists of a given ZK path and execute the action when it exists.
+ */
+ private void actOnExists(final String path, final Runnable action,
+ final SettableFuture<?> readyFuture, final long retryTime, final TimeUnit retryUnit) {
+ Futures.addCallback(zkClient.exists(path, new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ if (!isRunning()) {
+ return;
+ }
+ if (event.getType() == Event.EventType.NodeCreated) {
+ action.run();
+ }
+ }
+ }), new FutureCallback<Stat>() {
+ @Override
+ public void onSuccess(Stat result) {
+ if (result != null) {
+ action.run();
+ } else {
+ // If the node doesn't exists, treat it as ready. When the node becomes available later, data will be
+ // fetched by the watcher.
+ readyFuture.set(null);
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ // Retry the operation based on the retry time.
+ Thread retryThread = new Thread("zk-broker-service-retry") {
+ @Override
+ public void run() {
+ try {
+ retryUnit.sleep(retryTime);
+ actOnExists(path, action, readyFuture, retryTime, retryUnit);
+ } catch (InterruptedException e) {
+ LOG.warn("ZK retry thread interrupted. Action not retried.");
+ }
+ }
+ };
+ retryThread.setDaemon(true);
+ retryThread.start();
+ }
+ }, executorService);
+ }
+
+ /**
+ * Creates a supplier that always return latest copy from an {@link java.util.concurrent.atomic.AtomicReference}.
+ */
+ private <T> Supplier<T> createSupplier(final AtomicReference<T> ref) {
+ return new Supplier<T>() {
+ @Override
+ public T get() {
+ return ref.get();
+ }
+ };
+ }
+
+
+ /**
+ * Interface for invalidating an entry in a cache.
+ * @param <T> Key type.
+ */
+ private interface CacheInvalidater<T> {
+ void invalidate(T key);
+ }
+
+ /**
+ * Represents a path in zookeeper for cache key.
+ */
+ private interface KeyPath {
+ String getPath();
+ }
+
+ private static final class BrokerId implements KeyPath {
+ private final int id;
+
+ private BrokerId(int id) {
+ this.id = id;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return this == o || !(o == null || getClass() != o.getClass()) && id == ((BrokerId) o).id;
+ }
+
+ @Override
+ public int hashCode() {
+ return Ints.hashCode(id);
+ }
+
+ @Override
+ public String getPath() {
+ return BROKER_IDS_PATH + "/" + id;
+ }
+ }
+
+ /**
+ * Represents a topic + partition combination. Used for loading cache key.
+ */
+ private static final class KeyPathTopicPartition extends TopicPartition implements KeyPath {
+
+ private KeyPathTopicPartition(String topic, int partition) {
+ super(topic, partition);
+ }
+
+ @Override
+ public String getPath() {
+ return String.format("%s/%s/partitions/%d/state", BROKER_TOPICS_PATH, getTopic(), getPartition());
+ }
+ }
+
+ /**
+ * Class for holding information about a partition. Only used by gson to decode partition state node in zookeeper.
+ */
+ private static final class PartitionInfo {
+ private int[] isr;
+ private int leader;
+
+ private int[] getIsr() {
+ return isr;
+ }
+
+ private int getLeader() {
+ return leader;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/internal/kafka/client/ZKKafkaClientService.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/ZKKafkaClientService.java b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/ZKKafkaClientService.java
new file mode 100644
index 0000000..56f812b
--- /dev/null
+++ b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/ZKKafkaClientService.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.internal.kafka.client;
+
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.AbstractIdleService;
+import org.apache.twill.common.Cancellable;
+import org.apache.twill.common.Threads;
+import org.apache.twill.kafka.client.BrokerService;
+import org.apache.twill.kafka.client.Compression;
+import org.apache.twill.kafka.client.KafkaClientService;
+import org.apache.twill.kafka.client.KafkaConsumer;
+import org.apache.twill.kafka.client.KafkaPublisher;
+import org.apache.twill.zookeeper.ZKClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.ref.Reference;
+import java.lang.ref.ReferenceQueue;
+import java.lang.ref.WeakReference;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A KafkaClientService that uses ZooKeeper for broker discovery.
+ */
+public class ZKKafkaClientService extends AbstractIdleService implements KafkaClientService, Runnable {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ZKKafkaClientService.class);
+ private static final long PUBLISHER_CLEANUP_SECONDS = 1;
+
+ private final BrokerService brokerService;
+
+ // Maintains a weak reference key map for calling publisher.shutdown when garbage collected.
+ private final Map<WeakReference<KafkaPublisher>, Cancellable> publishers;
+ private final ReferenceQueue<KafkaPublisher> referenceQueue;
+
+ private final SimpleKafkaConsumer consumer;
+
+ // For running cleanup job
+ private ScheduledExecutorService scheduler;
+
+ public ZKKafkaClientService(ZKClient zkClient) {
+ this.brokerService = new ZKBrokerService(zkClient);
+ this.publishers = Collections.synchronizedMap(new IdentityHashMap<WeakReference<KafkaPublisher>, Cancellable>());
+ this.referenceQueue = new ReferenceQueue<KafkaPublisher>();
+ this.consumer = new SimpleKafkaConsumer(brokerService);
+ }
+
+ @Override
+ public KafkaPublisher getPublisher(KafkaPublisher.Ack ack, Compression compression) {
+ Preconditions.checkState(isRunning(), "Service is not running.");
+ String brokerList = brokerService.getBrokerList();
+ Preconditions.checkState(!brokerList.isEmpty(), "No broker available. Try again later.");
+
+ // Wrap the publisher with a weak reference and save the cancellable for closing the publisher.
+ SimpleKafkaPublisher publisher = new SimpleKafkaPublisher(brokerList, ack, compression);
+ publishers.put(new WeakReference<KafkaPublisher>(publisher, referenceQueue), publisher.start());
+ return publisher;
+ }
+
+ @Override
+ public KafkaConsumer getConsumer() {
+ Preconditions.checkState(isRunning(), "Service is not running.");
+ return consumer;
+ }
+
+ @Override
+ public void run() {
+ // For calling publisher.producer.close() on garbage collected
+ Reference<? extends KafkaPublisher> ref = referenceQueue.poll();
+ while (ref != null && isRunning()) {
+ publishers.remove(ref).cancel();
+ ref = referenceQueue.poll();
+ }
+ }
+
+ @Override
+ protected void startUp() throws Exception {
+ scheduler = Executors.newSingleThreadScheduledExecutor(Threads.createDaemonThreadFactory("kafka-client-cleanup"));
+ scheduler.scheduleAtFixedRate(this, PUBLISHER_CLEANUP_SECONDS, PUBLISHER_CLEANUP_SECONDS, TimeUnit.SECONDS);
+
+ // Start broker service to get auto-updated brokers information.
+ brokerService.startAndWait();
+ }
+
+ @Override
+ protected void shutDown() throws Exception {
+ LOG.info("Stopping KafkaClientService");
+ scheduler.shutdownNow();
+ for (Cancellable cancellable : publishers.values()) {
+ cancellable.cancel();
+ }
+ consumer.stop();
+
+ brokerService.stopAndWait();
+ LOG.info("KafkaClientService stopped");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/internal/kafka/client/package-info.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/package-info.java b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/package-info.java
index f3f615c..8676765 100644
--- a/twill-core/src/main/java/org/apache/twill/internal/kafka/client/package-info.java
+++ b/twill-core/src/main/java/org/apache/twill/internal/kafka/client/package-info.java
@@ -16,6 +16,6 @@
* limitations under the License.
*/
/**
- * This package provides pure java kafka client implementation.
+ * This package provides Kafka client implementation.
*/
package org.apache.twill.internal.kafka.client;
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/internal/logging/KafkaAppender.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/logging/KafkaAppender.java b/twill-core/src/main/java/org/apache/twill/internal/logging/KafkaAppender.java
index 12818ef..9a22bea 100644
--- a/twill-core/src/main/java/org/apache/twill/internal/logging/KafkaAppender.java
+++ b/twill-core/src/main/java/org/apache/twill/internal/logging/KafkaAppender.java
@@ -17,16 +17,6 @@
*/
package org.apache.twill.internal.logging;
-import org.apache.twill.common.Services;
-import org.apache.twill.common.Threads;
-import org.apache.twill.internal.kafka.client.Compression;
-import org.apache.twill.internal.kafka.client.SimpleKafkaClient;
-import org.apache.twill.kafka.client.KafkaClient;
-import org.apache.twill.kafka.client.PreparePublish;
-import org.apache.twill.zookeeper.RetryStrategies;
-import org.apache.twill.zookeeper.ZKClientService;
-import org.apache.twill.zookeeper.ZKClientServices;
-import org.apache.twill.zookeeper.ZKClients;
import ch.qos.logback.classic.pattern.ClassOfCallerConverter;
import ch.qos.logback.classic.pattern.FileOfCallerConverter;
import ch.qos.logback.classic.pattern.LineOfCallerConverter;
@@ -36,7 +26,6 @@ import ch.qos.logback.classic.spi.IThrowableProxy;
import ch.qos.logback.classic.spi.StackTraceElementProxy;
import ch.qos.logback.core.AppenderBase;
import com.google.common.base.Charsets;
-import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import com.google.common.collect.Iterables;
@@ -44,6 +33,16 @@ import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.gson.stream.JsonWriter;
+import org.apache.twill.common.Services;
+import org.apache.twill.common.Threads;
+import org.apache.twill.internal.kafka.client.ZKKafkaClientService;
+import org.apache.twill.kafka.client.Compression;
+import org.apache.twill.kafka.client.KafkaClientService;
+import org.apache.twill.kafka.client.KafkaPublisher;
+import org.apache.twill.zookeeper.RetryStrategies;
+import org.apache.twill.zookeeper.ZKClientService;
+import org.apache.twill.zookeeper.ZKClientServices;
+import org.apache.twill.zookeeper.ZKClients;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -65,7 +64,7 @@ public final class KafkaAppender extends AppenderBase<ILoggingEvent> {
private static final Logger LOG = LoggerFactory.getLogger(KafkaAppender.class);
private final LogEventConverter eventConverter;
- private final AtomicReference<PreparePublish> publisher;
+ private final AtomicReference<KafkaPublisher.Preparer> publisher;
private final Runnable flushTask;
/**
* Rough count of how many entries are being buffered. It's just approximate, not exact.
@@ -73,7 +72,7 @@ public final class KafkaAppender extends AppenderBase<ILoggingEvent> {
private final AtomicInteger bufferedSize;
private ZKClientService zkClientService;
- private KafkaClient kafkaClient;
+ private KafkaClientService kafkaClient;
private String zkConnectStr;
private String hostname;
private String topic;
@@ -84,7 +83,7 @@ public final class KafkaAppender extends AppenderBase<ILoggingEvent> {
public KafkaAppender() {
eventConverter = new LogEventConverter();
- publisher = new AtomicReference<PreparePublish>();
+ publisher = new AtomicReference<KafkaPublisher.Preparer>();
flushTask = createFlushTask();
bufferedSize = new AtomicInteger();
buffer = new ConcurrentLinkedQueue<String>();
@@ -141,12 +140,14 @@ public final class KafkaAppender extends AppenderBase<ILoggingEvent> {
ZKClients.retryOnFailure(ZKClientService.Builder.of(zkConnectStr).build(),
RetryStrategies.fixDelay(1, TimeUnit.SECONDS))));
- kafkaClient = new SimpleKafkaClient(zkClientService);
+ kafkaClient = new ZKKafkaClientService(zkClientService);
Futures.addCallback(Services.chainStart(zkClientService, kafkaClient), new FutureCallback<Object>() {
@Override
public void onSuccess(Object result) {
LOG.info("Kafka client started: " + zkConnectStr);
- publisher.set(kafkaClient.preparePublish(topic, Compression.SNAPPY));
+ KafkaPublisher.Preparer preparer = kafkaClient.getPublisher(KafkaPublisher.Ack.LEADER_RECEIVED,
+ Compression.SNAPPY).prepare(topic);
+ publisher.set(preparer);
scheduler.scheduleWithFixedDelay(flushTask, 0, flushPeriod, TimeUnit.MILLISECONDS);
}
@@ -186,7 +187,7 @@ public final class KafkaAppender extends AppenderBase<ILoggingEvent> {
private ListenableFuture<Integer> publishLogs() {
// If the publisher is not available, simply returns a completed future.
- PreparePublish publisher = KafkaAppender.this.publisher.get();
+ KafkaPublisher.Preparer publisher = KafkaAppender.this.publisher.get();
if (publisher == null) {
return Futures.immediateFuture(0);
}
@@ -202,13 +203,7 @@ public final class KafkaAppender extends AppenderBase<ILoggingEvent> {
}
bufferedSize.set(0);
- final int finalCount = count;
- return Futures.transform(publisher.publish(), new Function<Object, Integer>() {
- @Override
- public Integer apply(Object input) {
- return finalCount;
- }
- });
+ return publisher.send();
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/internal/logging/KafkaTwillRunnable.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/internal/logging/KafkaTwillRunnable.java b/twill-core/src/main/java/org/apache/twill/internal/logging/KafkaTwillRunnable.java
deleted file mode 100644
index c1695de..0000000
--- a/twill-core/src/main/java/org/apache/twill/internal/logging/KafkaTwillRunnable.java
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.twill.internal.logging;
-
-import org.apache.twill.api.Command;
-import org.apache.twill.api.TwillContext;
-import org.apache.twill.api.TwillRunnable;
-import org.apache.twill.api.TwillRunnableSpecification;
-import org.apache.twill.internal.EnvKeys;
-import org.apache.twill.internal.kafka.EmbeddedKafkaServer;
-import org.apache.twill.internal.utils.Networks;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Throwables;
-import com.google.common.collect.ImmutableMap;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.CountDownLatch;
-
-/**
- * A {@link org.apache.twill.api.TwillRunnable} for managing Kafka server.
- */
-public final class KafkaTwillRunnable implements TwillRunnable {
-
- private static final Logger LOG = LoggerFactory.getLogger(KafkaTwillRunnable.class);
-
- private final String kafkaDir;
- private EmbeddedKafkaServer server;
- private CountDownLatch stopLatch;
-
- public KafkaTwillRunnable(String kafkaDir) {
- this.kafkaDir = kafkaDir;
- }
-
- @Override
- public TwillRunnableSpecification configure() {
- return TwillRunnableSpecification.Builder.with()
- .setName("kafka")
- .withConfigs(ImmutableMap.of("kafkaDir", kafkaDir))
- .build();
- }
-
- @Override
- public void initialize(TwillContext context) {
- Map<String, String> args = context.getSpecification().getConfigs();
- String zkConnectStr = System.getenv(EnvKeys.TWILL_LOG_KAFKA_ZK);
- stopLatch = new CountDownLatch(1);
-
- try {
- server = new EmbeddedKafkaServer(new File(args.get("kafkaDir")), generateKafkaConfig(zkConnectStr));
- server.startAndWait();
- } catch (Exception e) {
- throw Throwables.propagate(e);
- }
- }
-
- @Override
- public void handleCommand(Command command) throws Exception {
- }
-
- @Override
- public void stop() {
- stopLatch.countDown();
- }
-
- @Override
- public void destroy() {
- server.stopAndWait();
- }
-
- @Override
- public void run() {
- try {
- stopLatch.await();
- } catch (InterruptedException e) {
- LOG.info("Running thread interrupted, shutting down kafka server.", e);
- }
- }
-
- private Properties generateKafkaConfig(String zkConnectStr) {
- int port = Networks.getRandomPort();
- Preconditions.checkState(port > 0, "Failed to get random port.");
-
- Properties prop = new Properties();
- prop.setProperty("log.dir", new File("kafka-logs").getAbsolutePath());
- prop.setProperty("zk.connect", zkConnectStr);
- prop.setProperty("num.threads", "8");
- prop.setProperty("port", Integer.toString(port));
- prop.setProperty("log.flush.interval", "10000");
- prop.setProperty("max.socket.request.bytes", "104857600");
- prop.setProperty("log.cleanup.interval.mins", "1");
- prop.setProperty("log.default.flush.scheduler.interval.ms", "1000");
- prop.setProperty("zk.connectiontimeout.ms", "1000000");
- prop.setProperty("socket.receive.buffer", "1048576");
- prop.setProperty("enable.zookeeper", "true");
- prop.setProperty("log.retention.hours", "168");
- prop.setProperty("brokerid", "0");
- prop.setProperty("socket.send.buffer", "1048576");
- prop.setProperty("num.partitions", "1");
- prop.setProperty("log.file.size", "536870912");
- prop.setProperty("log.default.flush.interval.ms", "1000");
- return prop;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/kafka/client/BrokerInfo.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/kafka/client/BrokerInfo.java b/twill-core/src/main/java/org/apache/twill/kafka/client/BrokerInfo.java
new file mode 100644
index 0000000..e659ab7
--- /dev/null
+++ b/twill-core/src/main/java/org/apache/twill/kafka/client/BrokerInfo.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.kafka.client;
+
+import com.google.common.base.Objects;
+
+/**
+ * Represents broker information. This class should be serializable with Gson.
+ */
+public final class BrokerInfo {
+
+ private String host;
+ private int port;
+
+ public String getHost() {
+ return host;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ BrokerInfo that = (BrokerInfo) o;
+ return host.equals(that.getHost()) && port == that.getPort();
+ }
+
+ @Override
+ public int hashCode() {
+ int result = host.hashCode();
+ result = 31 * result + port;
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return Objects.toStringHelper(BrokerInfo.class)
+ .add("host", host)
+ .add("port", port)
+ .toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/kafka/client/BrokerService.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/kafka/client/BrokerService.java b/twill-core/src/main/java/org/apache/twill/kafka/client/BrokerService.java
new file mode 100644
index 0000000..97270bd
--- /dev/null
+++ b/twill-core/src/main/java/org/apache/twill/kafka/client/BrokerService.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.kafka.client;
+
+import com.google.common.util.concurrent.Service;
+
+/**
+ * Service for providing information of kafka brokers.
+ */
+public interface BrokerService extends Service {
+
+ /**
+ * Returns the broker information of the current leader of the given topic and partition.
+ * @param topic Topic for looking up for leader.
+ * @param partition Partition for looking up for leader.
+ * @return A {@link BrokerInfo} containing information about the current leader, or {@code null} if
+ * current leader is unknown.
+ */
+ BrokerInfo getLeader(String topic, int partition);
+
+ /**
+ * Returns a live iterable that gives information for all the known brokers.
+ * @return An {@link Iterable} of {@link BrokerInfo} that when {@link Iterable#iterator()} is called, it returns
+ * an iterator that gives the latest list of {@link BrokerInfo}.
+ */
+ Iterable<BrokerInfo> getBrokers();
+
+ /**
+ * Returns a comma separate string of all current brokers.
+ * @return A string in the format {@code host1:port1,host2:port2} or empty string if no broker has been discovered.
+ */
+ String getBrokerList();
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/kafka/client/Compression.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/kafka/client/Compression.java b/twill-core/src/main/java/org/apache/twill/kafka/client/Compression.java
new file mode 100644
index 0000000..8597d3b
--- /dev/null
+++ b/twill-core/src/main/java/org/apache/twill/kafka/client/Compression.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.kafka.client;
+
+/**
+ * An Enum representing different compression codec options for publishing messages.
+ */
+public enum Compression {
+
+ NONE,
+ GZIP,
+ SNAPPY;
+
+ private final String codec;
+
+ private Compression() {
+ this.codec = name().toLowerCase();
+ }
+
+ public String getCodec() {
+ return codec;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/kafka/client/FetchException.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/kafka/client/FetchException.java b/twill-core/src/main/java/org/apache/twill/kafka/client/FetchException.java
deleted file mode 100644
index acccf04..0000000
--- a/twill-core/src/main/java/org/apache/twill/kafka/client/FetchException.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.twill.kafka.client;
-
-/**
- *
- */
-public final class FetchException extends RuntimeException {
-
- private final ErrorCode errorCode;
-
- public FetchException(String message, ErrorCode errorCode) {
- super(message);
- this.errorCode = errorCode;
- }
-
- public ErrorCode getErrorCode() {
- return errorCode;
- }
-
- @Override
- public String toString() {
- return String.format("%s. Error code: %s", super.toString(), errorCode);
- }
-
- public enum ErrorCode {
- UNKNOWN(-1),
- OK(0),
- OFFSET_OUT_OF_RANGE(1),
- INVALID_MESSAGE(2),
- WRONG_PARTITION(3),
- INVALID_FETCH_SIZE(4);
-
- private final int code;
-
- ErrorCode(int code) {
- this.code = code;
- }
-
- public int getCode() {
- return code;
- }
-
- public static ErrorCode fromCode(int code) {
- switch (code) {
- case -1:
- return UNKNOWN;
- case 0:
- return OK;
- case 1:
- return OFFSET_OUT_OF_RANGE;
- case 2:
- return INVALID_MESSAGE;
- case 3:
- return WRONG_PARTITION;
- case 4:
- return INVALID_FETCH_SIZE;
- }
- throw new IllegalArgumentException("Unknown error code");
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/kafka/client/FetchedMessage.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/kafka/client/FetchedMessage.java b/twill-core/src/main/java/org/apache/twill/kafka/client/FetchedMessage.java
index 65e140f..5739ac6 100644
--- a/twill-core/src/main/java/org/apache/twill/kafka/client/FetchedMessage.java
+++ b/twill-core/src/main/java/org/apache/twill/kafka/client/FetchedMessage.java
@@ -20,17 +20,16 @@ package org.apache.twill.kafka.client;
import java.nio.ByteBuffer;
/**
- * Represents a message fetched from kafka broker.
+ *
*/
public interface FetchedMessage {
- /**
- * Returns the message offset.
- */
- long getOffset();
+ TopicPartition getTopicPartition();
+
+ ByteBuffer getPayload();
/**
- * Returns the message payload.
+ * Returns the offset for the next message to be read.
*/
- ByteBuffer getBuffer();
+ long getNextOffset();
}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/kafka/client/KafkaClient.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/kafka/client/KafkaClient.java b/twill-core/src/main/java/org/apache/twill/kafka/client/KafkaClient.java
index 496195b..f98d879 100644
--- a/twill-core/src/main/java/org/apache/twill/kafka/client/KafkaClient.java
+++ b/twill-core/src/main/java/org/apache/twill/kafka/client/KafkaClient.java
@@ -17,34 +17,22 @@
*/
package org.apache.twill.kafka.client;
-import org.apache.twill.internal.kafka.client.Compression;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.Service;
-
-import java.util.Iterator;
-
/**
- * This interface provides methods for interacting with kafka broker. It also
- * extends from {@link Service} for lifecycle management. The {@link #start()} method
- * must be called prior to other methods in this class. When instance of this class
- * is not needed, call {@link #stop()}} to release any resources that it holds.
+ * Represents a kafka client that can publish/subscribe to a Kafka server cluster.
*/
-public interface KafkaClient extends Service {
-
- PreparePublish preparePublish(String topic, Compression compression);
+public interface KafkaClient {
- Iterator<FetchedMessage> consume(String topic, int partition, long offset, int maxSize);
+ /**
+ * Creates a {@link KafkaPublisher} that is ready for publish.
+ * @param ack Type of ack that the publisher would use for all it's publish.
+ * @param compression The compression type for messages published through the returned publisher.
+ * @return A {@link KafkaPublisher}.
+ */
+ KafkaPublisher getPublisher(KafkaPublisher.Ack ack, Compression compression);
/**
- * Fetches offset from the given topic and partition.
- * @param topic Topic to fetch from.
- * @param partition Partition to fetch from.
- * @param time The first offset of every segment file for a given partition with a modified time less than time.
- * {@code -1} for latest offset, {@code -2} for earliest offset.
- * @param maxOffsets Maximum number of offsets to fetch.
- * @return A Future that carry the result as an array of offsets in descending order.
- * The size of the result array would not be larger than maxOffsets. If there is any error during the fetch,
- * the exception will be carried in the exception.
+ * Creates a {@link KafkaConsumer} for consuming messages.
+ * @return A {@link KafkaConsumer}.
*/
- ListenableFuture<long[]> getOffset(String topic, int partition, long time, int maxOffsets);
+ KafkaConsumer getConsumer();
}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/fcbe54d5/twill-core/src/main/java/org/apache/twill/kafka/client/KafkaClientService.java
----------------------------------------------------------------------
diff --git a/twill-core/src/main/java/org/apache/twill/kafka/client/KafkaClientService.java b/twill-core/src/main/java/org/apache/twill/kafka/client/KafkaClientService.java
new file mode 100644
index 0000000..d9224e0
--- /dev/null
+++ b/twill-core/src/main/java/org/apache/twill/kafka/client/KafkaClientService.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.kafka.client;
+
+import com.google.common.util.concurrent.Service;
+
+/**
+ * A {@link KafkaClient} that extends {@link Service} to provide lifecycle management.
+ */
+public interface KafkaClientService extends KafkaClient, Service {
+}