clientProperties)
- {
- super.setClientProperties(clientProperties);
+ int result = host != null ? host.hashCode() : 0;
+ result = 31 * result + port;
+ result = 31 * result + (username != null ? username.hashCode() : 0);
+ result = 31 * result + (password != null ? password.hashCode() : 0);
+ result = 31 * result + (virtualHost != null ? virtualHost.hashCode() : 0);
+ result = 31 * result + (uri != null ? uri.hashCode() : 0);
+ result = 31 * result + requestedChannelMax;
+ result = 31 * result + requestedFrameMax;
+ result = 31 * result + requestedHeartbeat;
+ result = 31 * result + connectionTimeout;
+ result = 31 * result + (clientProperties != null ? clientProperties.hashCode() : 0);
+ return result;
}
}
diff --git a/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQDruidModule.java b/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQDruidModule.java
index 548cbcc1d1a..8ca79bbea99 100644
--- a/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQDruidModule.java
+++ b/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQDruidModule.java
@@ -29,7 +29,7 @@ import java.util.List;
/**
*/
-public class RabbitMQDruidModule implements DruidModule
+public class RabbitMQDruidModule implements DruidModule
{
@Override
public List getJacksonModules()
diff --git a/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQFirehoseConfig.java b/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQFirehoseConfig.java
index 7bae291c8a3..325bc41cd92 100644
--- a/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQFirehoseConfig.java
+++ b/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQFirehoseConfig.java
@@ -19,6 +19,7 @@
package io.druid.firehose.rabbitmq;
+import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
/**
@@ -26,17 +27,50 @@ import com.fasterxml.jackson.annotation.JsonProperty;
*/
public class RabbitMQFirehoseConfig
{
- private String queue = null;
- private String exchange = null;
- private String routingKey = null;
- private boolean durable = false;
- private boolean exclusive = false;
- private boolean autoDelete = false;
-
// Lyra (auto reconnect) properties
- private int maxRetries = 100;
- private int retryIntervalSeconds = 2;
- private long maxDurationSeconds = 5 * 60;
+ private static final int defaultMaxRetries = 100;
+ private static final int defaultRetryIntervalSeconds = 2;
+ private static final long defaultMaxDurationSeconds = 5 * 60;
+
+ public static RabbitMQFirehoseConfig makeDefaultConfig()
+ {
+ return new RabbitMQFirehoseConfig(null, null, null, false, false, false, 0, 0, 0);
+ }
+
+ private final String queue;
+ private final String exchange;
+ private final String routingKey;
+ private final boolean durable;
+ private final boolean exclusive;
+ private final boolean autoDelete;
+ private final int maxRetries;
+ private final int retryIntervalSeconds;
+ private final long maxDurationSeconds;
+
+ @JsonCreator
+ public RabbitMQFirehoseConfig(
+ @JsonProperty("queue") String queue,
+ @JsonProperty("exchange") String exchange,
+ @JsonProperty("routingKey") String routingKey,
+ @JsonProperty("durable") boolean durable,
+ @JsonProperty("exclusive") boolean exclusive,
+ @JsonProperty("autoDelete") boolean autoDelete,
+ @JsonProperty("maxRetries") int maxRetries,
+ @JsonProperty("retryIntervalSeconds") int retryIntervalSeconds,
+ @JsonProperty("maxDurationSeconds") long maxDurationSeconds
+ )
+ {
+ this.queue = queue;
+ this.exchange = exchange;
+ this.routingKey = routingKey;
+ this.durable = durable;
+ this.exclusive = exclusive;
+ this.autoDelete = autoDelete;
+
+ this.maxRetries = maxRetries == 0 ? defaultMaxRetries : maxRetries;
+ this.retryIntervalSeconds = retryIntervalSeconds == 0 ? defaultRetryIntervalSeconds : retryIntervalSeconds;
+ this.maxDurationSeconds = maxDurationSeconds == 0 ? defaultMaxDurationSeconds : maxDurationSeconds;
+ }
@JsonProperty
public String getQueue()
@@ -44,90 +78,109 @@ public class RabbitMQFirehoseConfig
return queue;
}
- public void setQueue(String queue)
- {
- this.queue = queue;
- }
-
@JsonProperty
public String getExchange()
{
return exchange;
}
- public void setExchange(String exchange)
- {
- this.exchange = exchange;
- }
-
@JsonProperty
public String getRoutingKey()
{
return routingKey;
}
- public void setRoutingKey(String routingKey)
- {
- this.routingKey = routingKey;
- }
-
@JsonProperty
public boolean isDurable()
{
return durable;
}
- public void setDurable(boolean durable)
- {
- this.durable = durable;
- }
-
@JsonProperty
public boolean isExclusive()
{
return exclusive;
}
- public void setExclusive(boolean exclusive)
- {
- this.exclusive = exclusive;
- }
-
@JsonProperty
public boolean isAutoDelete()
{
return autoDelete;
}
- public void setAutoDelete(boolean autoDelete)
- {
- this.autoDelete = autoDelete;
- }
-
@JsonProperty
- public int getMaxRetries() {
+ public int getMaxRetries()
+ {
return maxRetries;
}
- public void setMaxRetries(int maxRetries) {
- this.maxRetries = maxRetries;
- }
-
@JsonProperty
- public int getRetryIntervalSeconds() {
+ public int getRetryIntervalSeconds()
+ {
return retryIntervalSeconds;
}
- public void setRetryIntervalSeconds(int retryIntervalSeconds) {
- this.retryIntervalSeconds = retryIntervalSeconds;
- }
-
@JsonProperty
- public long getMaxDurationSeconds() {
+ public long getMaxDurationSeconds()
+ {
return maxDurationSeconds;
}
- public void setMaxDurationSeconds(int maxDurationSeconds) {
- this.maxDurationSeconds = maxDurationSeconds;
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ RabbitMQFirehoseConfig that = (RabbitMQFirehoseConfig) o;
+
+ if (autoDelete != that.autoDelete) {
+ return false;
+ }
+ if (durable != that.durable) {
+ return false;
+ }
+ if (exclusive != that.exclusive) {
+ return false;
+ }
+ if (maxDurationSeconds != that.maxDurationSeconds) {
+ return false;
+ }
+ if (maxRetries != that.maxRetries) {
+ return false;
+ }
+ if (retryIntervalSeconds != that.retryIntervalSeconds) {
+ return false;
+ }
+ if (exchange != null ? !exchange.equals(that.exchange) : that.exchange != null) {
+ return false;
+ }
+ if (queue != null ? !queue.equals(that.queue) : that.queue != null) {
+ return false;
+ }
+ if (routingKey != null ? !routingKey.equals(that.routingKey) : that.routingKey != null) {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ int result = queue != null ? queue.hashCode() : 0;
+ result = 31 * result + (exchange != null ? exchange.hashCode() : 0);
+ result = 31 * result + (routingKey != null ? routingKey.hashCode() : 0);
+ result = 31 * result + (durable ? 1 : 0);
+ result = 31 * result + (exclusive ? 1 : 0);
+ result = 31 * result + (autoDelete ? 1 : 0);
+ result = 31 * result + maxRetries;
+ result = 31 * result + retryIntervalSeconds;
+ result = 31 * result + (int) (maxDurationSeconds ^ (maxDurationSeconds >>> 32));
+ return result;
}
}
diff --git a/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQFirehoseFactory.java b/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQFirehoseFactory.java
index 8504c70ed9d..f38398bcc84 100644
--- a/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQFirehoseFactory.java
+++ b/rabbitmq/src/main/java/io/druid/firehose/rabbitmq/RabbitMQFirehoseFactory.java
@@ -26,7 +26,6 @@ import com.metamx.common.logger.Logger;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
@@ -50,14 +49,14 @@ import java.util.concurrent.LinkedBlockingQueue;
/**
* A FirehoseFactory for RabbitMQ.
- *
+ *
* It will receive it's configuration through the realtime.spec file and expects to find a
* consumerProps element in the firehose definition with values for a number of configuration options.
* Below is a complete example for a RabbitMQ firehose configuration with some explanation. Options
* that have defaults can be skipped but options with no defaults must be specified with the exception
* of the URI property. If the URI property is set, it will override any other property that was also
* set.
- *
+ *
* File: realtime.spec
*
* "firehose" : {
@@ -89,7 +88,7 @@ import java.util.concurrent.LinkedBlockingQueue;
* }
* },
*
- *
+ *
* Limitations: This implementation will not attempt to reconnect to the MQ broker if the
* connection to it is lost. Furthermore it does not support any automatic failover on high availability
* RabbitMQ clusters. This is not supported by the underlying AMQP client library and while the behavior
@@ -97,7 +96,7 @@ import java.util.concurrent.LinkedBlockingQueue;
* the RabbitMQ cluster that sets the "ha-mode" and "ha-sync-mode" properly on the queue that this
* Firehose connects to, messages should survive an MQ broker node failure and be delivered once a
* connection to another node is set up.
- *
+ *
* For more information on RabbitMQ high availability please see:
* http://www.rabbitmq.com/ha.html.
*/
@@ -105,27 +104,36 @@ public class RabbitMQFirehoseFactory implements FirehoseFactory _queue;
- public QueueingConsumer(Channel ch) {
+ public QueueingConsumer(Channel ch)
+ {
this(ch, new LinkedBlockingQueue());
}
- public QueueingConsumer(Channel ch, BlockingQueue q) {
+ public QueueingConsumer(Channel ch, BlockingQueue q)
+ {
super(ch);
this._queue = q;
}
- @Override public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
+ @Override
+ public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig)
+ {
_queue.clear();
}
- @Override public void handleCancel(String consumerTag) throws IOException {
+ @Override
+ public void handleCancel(String consumerTag) throws IOException
+ {
_queue.clear();
}
- @Override public void handleDelivery(String consumerTag,
- Envelope envelope,
- AMQP.BasicProperties properties,
- byte[] body)
- throws IOException
+ @Override
+ public void handleDelivery(
+ String consumerTag,
+ Envelope envelope,
+ AMQP.BasicProperties properties,
+ byte[] body
+ )
+ throws IOException
{
this._queue.add(new Delivery(envelope, properties, body));
}
public Delivery nextDelivery()
- throws InterruptedException, ShutdownSignalException, ConsumerCancelledException
+ throws InterruptedException, ShutdownSignalException, ConsumerCancelledException
{
return _queue.take();
}
diff --git a/rabbitmq/src/test/java/io/druid/examples/rabbitmq/RabbitMQFirehoseFactoryTest.java b/rabbitmq/src/test/java/io/druid/examples/rabbitmq/RabbitMQFirehoseFactoryTest.java
new file mode 100644
index 00000000000..eca04717fc2
--- /dev/null
+++ b/rabbitmq/src/test/java/io/druid/examples/rabbitmq/RabbitMQFirehoseFactoryTest.java
@@ -0,0 +1,139 @@
+/*
+ * Druid - a distributed column store.
+ * Copyright (C) 2012, 2013 Metamarkets Group Inc.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+ */
+
+package io.druid.examples.rabbitmq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import com.rabbitmq.client.ConnectionFactory;
+import io.druid.data.input.impl.DimensionsSpec;
+import io.druid.data.input.impl.JSONParseSpec;
+import io.druid.data.input.impl.SpatialDimensionSchema;
+import io.druid.data.input.impl.StringInputRowParser;
+import io.druid.data.input.impl.TimestampSpec;
+import io.druid.firehose.rabbitmq.JacksonifiedConnectionFactory;
+import io.druid.firehose.rabbitmq.RabbitMQFirehoseConfig;
+import io.druid.firehose.rabbitmq.RabbitMQFirehoseFactory;
+import io.druid.jackson.DefaultObjectMapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+
+/**
+ */
+public class RabbitMQFirehoseFactoryTest
+{
+ private static final ObjectMapper mapper = new DefaultObjectMapper();
+
+ @Test
+ public void testSerde() throws Exception
+ {
+ RabbitMQFirehoseConfig config = new RabbitMQFirehoseConfig(
+ "test",
+ "test2",
+ "test3",
+ true,
+ true,
+ true,
+ 5,
+ 10,
+ 20
+ );
+
+ JacksonifiedConnectionFactory connectionFactory = new JacksonifiedConnectionFactory(
+ "foo",
+ 9978,
+ "user",
+ "pw",
+ "host",
+ null,
+ 5,
+ 10,
+ 11,
+ 12,
+ ImmutableMap.of("hi", "bye")
+ );
+
+ RabbitMQFirehoseFactory factory = new RabbitMQFirehoseFactory(
+ connectionFactory,
+ config,
+ new StringInputRowParser(
+ new JSONParseSpec(
+ new TimestampSpec("timestamp", "auto"),
+ new DimensionsSpec(
+ Arrays.asList("dim"),
+ Lists.newArrayList(),
+ Lists.newArrayList()
+ )
+ ),
+ null, null, null, null
+ )
+ );
+
+ byte[] bytes = mapper.writeValueAsBytes(factory);
+ RabbitMQFirehoseFactory factory2 = mapper.readValue(bytes, RabbitMQFirehoseFactory.class);
+ byte[] bytes2 = mapper.writeValueAsBytes(factory2);
+
+ Assert.assertArrayEquals(bytes, bytes2);
+
+ Assert.assertEquals(factory.getConfig(), factory2.getConfig());
+ Assert.assertEquals(factory.getConnectionFactory(), factory2.getConnectionFactory());
+ }
+
+ @Test
+ public void testDefaultSerde() throws Exception
+ {
+ RabbitMQFirehoseConfig config = RabbitMQFirehoseConfig.makeDefaultConfig();
+
+ JacksonifiedConnectionFactory connectionFactory = JacksonifiedConnectionFactory.makeDefaultConnectionFactory();
+
+ RabbitMQFirehoseFactory factory = new RabbitMQFirehoseFactory(
+ connectionFactory,
+ config,
+ new StringInputRowParser(
+ new JSONParseSpec(
+ new TimestampSpec("timestamp", "auto"),
+ new DimensionsSpec(
+ Arrays.asList("dim"),
+ Lists.newArrayList(),
+ Lists.newArrayList()
+ )
+ ),
+ null, null, null, null
+ )
+ );
+
+ byte[] bytes = mapper.writeValueAsBytes(factory);
+ RabbitMQFirehoseFactory factory2 = mapper.readValue(bytes, RabbitMQFirehoseFactory.class);
+ byte[] bytes2 = mapper.writeValueAsBytes(factory2);
+
+ Assert.assertArrayEquals(bytes, bytes2);
+
+ Assert.assertEquals(factory.getConfig(), factory2.getConfig());
+ Assert.assertEquals(factory.getConnectionFactory(), factory2.getConnectionFactory());
+
+ Assert.assertEquals(300, factory2.getConfig().getMaxDurationSeconds());
+
+ Assert.assertEquals(ConnectionFactory.DEFAULT_HOST, factory2.getConnectionFactory().getHost());
+ Assert.assertEquals(ConnectionFactory.DEFAULT_USER, factory2.getConnectionFactory().getUsername());
+ Assert.assertEquals(ConnectionFactory.DEFAULT_AMQP_PORT, factory2.getConnectionFactory().getPort());
+ }
+}
diff --git a/s3-extensions/pom.xml b/s3-extensions/pom.xml
index a512b649ecd..e4914713493 100644
--- a/s3-extensions/pom.xml
+++ b/s3-extensions/pom.xml
@@ -28,7 +28,7 @@
io.druid
druid
- 0.6.157-SNAPSHOT
+ 0.6.160-SNAPSHOT
diff --git a/server/pom.xml b/server/pom.xml
index 7382aefb237..740e06d56ad 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -28,7 +28,7 @@
io.druid
druid
- 0.6.157-SNAPSHOT
+ 0.6.160-SNAPSHOT
@@ -77,6 +77,10 @@
com.fasterxml.jackson.jaxrs
jackson-jaxrs-json-provider
+
+ com.fasterxml.jackson.jaxrs
+ jackson-jaxrs-smile-provider
+
com.fasterxml.jackson.dataformat
jackson-dataformat-smile
diff --git a/server/src/main/java/io/druid/client/CachingClusteredClient.java b/server/src/main/java/io/druid/client/CachingClusteredClient.java
index d0135eb7f4a..f189e060670 100644
--- a/server/src/main/java/io/druid/client/CachingClusteredClient.java
+++ b/server/src/main/java/io/druid/client/CachingClusteredClient.java
@@ -25,7 +25,6 @@ import com.google.common.base.Function;
import com.google.common.base.Supplier;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Iterables;
import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -261,14 +260,22 @@ public class CachingClusteredClient implements QueryRunner
Ordering.natural().onResultOf(Pair.>lhsFn())
);
- final Sequence> seq = Sequences.simple(
- Iterables.transform(listOfSequences, Pair.>rhsFn())
- );
- if (strategy == null) {
- return toolChest.mergeSequences(seq);
- } else {
- return strategy.mergeSequences(seq);
+ final List> orderedSequences = Lists.newLinkedList();
+ DateTime unorderedStart = null;
+ List> unordered = Lists.newLinkedList();
+ for (Pair> sequencePair : listOfSequences) {
+ if (unorderedStart != null && unorderedStart.getMillis() != sequencePair.lhs.getMillis()) {
+ orderedSequences.add(toolChest.mergeSequencesUnordered(Sequences.simple(unordered)));
+ unordered = Lists.newLinkedList();
+ }
+ unorderedStart = sequencePair.lhs;
+ unordered.add(sequencePair.rhs);
}
+ if(!unordered.isEmpty()) {
+ orderedSequences.add(toolChest.mergeSequencesUnordered(Sequences.simple(unordered)));
+ }
+
+ return toolChest.mergeSequences(Sequences.simple(orderedSequences));
}
private void addSequencesFromCache(ArrayList>> listOfSequences)
@@ -332,7 +339,9 @@ public class CachingClusteredClient implements QueryRunner
if (!server.isAssignable() || !populateCache || isBySegment) {
resultSeqToAdd = clientQueryable.run(query.withQuerySegmentSpec(segmentSpec));
} else {
- resultSeqToAdd = toolChest.mergeSequences(
+ // this could be more efficient, since we only need to reorder results
+ // for batches of segments with the same segment start time.
+ resultSeqToAdd = toolChest.mergeSequencesUnordered(
Sequences.map(
clientQueryable.run(rewrittenQuery.withQuerySegmentSpec(segmentSpec)),
new Function>()
diff --git a/server/src/main/java/io/druid/server/coordination/BaseZkCoordinator.java b/server/src/main/java/io/druid/server/coordination/BaseZkCoordinator.java
index f31cfb6311c..43776480829 100644
--- a/server/src/main/java/io/druid/server/coordination/BaseZkCoordinator.java
+++ b/server/src/main/java/io/druid/server/coordination/BaseZkCoordinator.java
@@ -37,7 +37,6 @@ import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.utils.ZKPaths;
import java.io.IOException;
-import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
@@ -212,6 +211,11 @@ public abstract class BaseZkCoordinator implements DataSegmentChangeHandler
}
}
+ public boolean isStarted()
+ {
+ return started;
+ }
+
public abstract void loadLocalCache();
public abstract DataSegmentChangeHandler getDataSegmentChangeHandler();
diff --git a/server/src/main/java/io/druid/server/coordination/ZkCoordinator.java b/server/src/main/java/io/druid/server/coordination/ZkCoordinator.java
index 34a0ade4574..ba139e8a370 100644
--- a/server/src/main/java/io/druid/server/coordination/ZkCoordinator.java
+++ b/server/src/main/java/io/druid/server/coordination/ZkCoordinator.java
@@ -169,7 +169,7 @@ public class ZkCoordinator extends BaseZkCoordinator
catch (IOException e) {
throw new SegmentLoadingException(e, "Failed to announce segment[%s]", segment.getIdentifier());
}
- };
+ }
}
catch (SegmentLoadingException e) {
log.makeAlert(e, "Failed to load segment for dataSource")
diff --git a/server/src/main/java/io/druid/server/http/HistoricalResource.java b/server/src/main/java/io/druid/server/http/HistoricalResource.java
new file mode 100644
index 00000000000..a4f5956a0cc
--- /dev/null
+++ b/server/src/main/java/io/druid/server/http/HistoricalResource.java
@@ -0,0 +1,32 @@
+package io.druid.server.http;
+
+import com.google.common.collect.ImmutableMap;
+import io.druid.server.coordination.ZkCoordinator;
+
+import javax.inject.Inject;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Response;
+
+@Path("/druid/historical/v1")
+public class HistoricalResource
+{
+ private final ZkCoordinator coordinator;
+
+ @Inject
+ public HistoricalResource(
+ ZkCoordinator coordinator
+ )
+ {
+ this.coordinator = coordinator;
+ }
+
+ @GET
+ @Path("/loadstatus")
+ @Produces("application/json")
+ public Response getLoadStatus()
+ {
+ return Response.ok(ImmutableMap.of("cacheInitialized", coordinator.isStarted())).build();
+ }
+}
diff --git a/server/src/main/java/io/druid/server/router/JavaScriptTieredBrokerSelectorStrategy.java b/server/src/main/java/io/druid/server/router/JavaScriptTieredBrokerSelectorStrategy.java
new file mode 100644
index 00000000000..afc7997de43
--- /dev/null
+++ b/server/src/main/java/io/druid/server/router/JavaScriptTieredBrokerSelectorStrategy.java
@@ -0,0 +1,106 @@
+/*
+ * Druid - a distributed column store.
+ * Copyright (C) 2014 Metamarkets Group Inc.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+ */
+
+package io.druid.server.router;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
+import io.druid.query.Query;
+
+import javax.script.Compilable;
+import javax.script.Invocable;
+import javax.script.ScriptEngine;
+import javax.script.ScriptEngineManager;
+import javax.script.ScriptException;
+
+public class JavaScriptTieredBrokerSelectorStrategy implements TieredBrokerSelectorStrategy
+{
+ public static interface SelectorFunction
+ {
+ public String apply(TieredBrokerConfig config, Query query);
+ }
+
+ private final SelectorFunction fnSelector;
+ private final String function;
+
+ @JsonCreator
+ public JavaScriptTieredBrokerSelectorStrategy(@JsonProperty("function") String fn)
+ {
+ Preconditions.checkNotNull(fn, "function must not be null");
+
+ final ScriptEngine engine = new ScriptEngineManager().getEngineByName("javascript");
+ try {
+ ((Compilable)engine).compile("var apply = " + fn).eval();
+ } catch(ScriptException e) {
+ Throwables.propagate(e);
+ }
+ this.function = fn;
+ this.fnSelector = ((Invocable)engine).getInterface(SelectorFunction.class);
+ }
+
+ @Override
+ public Optional getBrokerServiceName(
+ TieredBrokerConfig config, Query query
+ )
+ {
+ return Optional.fromNullable(fnSelector.apply(config, query));
+ }
+
+ @JsonProperty
+ public String getFunction()
+ {
+ return function;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ JavaScriptTieredBrokerSelectorStrategy that = (JavaScriptTieredBrokerSelectorStrategy) o;
+
+ if (!function.equals(that.function)) {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return function.hashCode();
+ }
+
+ @Override
+ public String toString()
+ {
+ return "JavascriptTieredBrokerSelectorStrategy{" +
+ "function='" + function + '\'' +
+ '}';
+ }
+}
diff --git a/server/src/main/java/io/druid/server/router/TieredBrokerSelectorStrategy.java b/server/src/main/java/io/druid/server/router/TieredBrokerSelectorStrategy.java
index 40a7714d870..10bffe4fb37 100644
--- a/server/src/main/java/io/druid/server/router/TieredBrokerSelectorStrategy.java
+++ b/server/src/main/java/io/druid/server/router/TieredBrokerSelectorStrategy.java
@@ -29,7 +29,8 @@ import io.druid.query.Query;
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
@JsonSubTypes(value = {
@JsonSubTypes.Type(name = "timeBoundary", value = TimeBoundaryTieredBrokerSelectorStrategy.class),
- @JsonSubTypes.Type(name = "priority", value = PriorityTieredBrokerSelectorStrategy.class)
+ @JsonSubTypes.Type(name = "priority", value = PriorityTieredBrokerSelectorStrategy.class),
+ @JsonSubTypes.Type(name = "javascript", value = JavaScriptTieredBrokerSelectorStrategy.class)
})
public interface TieredBrokerSelectorStrategy
diff --git a/server/src/main/java/io/druid/timeline/partition/HashBasedNumberedShardSpec.java b/server/src/main/java/io/druid/timeline/partition/HashBasedNumberedShardSpec.java
index be640a03545..56ff5159c62 100644
--- a/server/src/main/java/io/druid/timeline/partition/HashBasedNumberedShardSpec.java
+++ b/server/src/main/java/io/druid/timeline/partition/HashBasedNumberedShardSpec.java
@@ -49,14 +49,14 @@ public class HashBasedNumberedShardSpec extends NumberedShardSpec
}
@Override
- public boolean isInChunk(InputRow inputRow)
+ public boolean isInChunk(long timestamp, InputRow inputRow)
{
- return (((long) hash(inputRow)) - getPartitionNum()) % getPartitions() == 0;
+ return (((long) hash(timestamp, inputRow)) - getPartitionNum()) % getPartitions() == 0;
}
- protected int hash(InputRow inputRow)
+ protected int hash(long timestamp, InputRow inputRow)
{
- final List