storm-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kabh...@apache.org
Subject [05/13] storm git commit: Implement EsLookupBolt
Date Thu, 27 Aug 2015 22:12:49 GMT
Implement EsLookupBolt


Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/59999924
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/59999924
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/59999924

Branch: refs/heads/master
Commit: 599999246f4ce251d44c50a64423ccf8ea464b34
Parents: 3094baf
Author: Alex Panov <alex.panov@teradata.com>
Authored: Thu Aug 20 12:30:17 2015 +0200
Committer: Alex Panov <alex.panov@teradata.com>
Committed: Thu Aug 20 12:30:17 2015 +0200

----------------------------------------------------------------------
 .../elasticsearch/ElasticsearchGetRequest.java  |  10 ++
 .../storm/elasticsearch/bolt/EsLookupBolt.java  |  70 ++++++++++
 .../bolt/EsLookupResultOutput.java              |  15 +++
 .../bolt/EsLookupBoltIntegrationTest.java       | 130 +++++++++++++++++++
 .../elasticsearch/bolt/EsLookupBoltTest.java    | 102 +++++++++++++++
 5 files changed, 327 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/storm/blob/59999924/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/ElasticsearchGetRequest.java
----------------------------------------------------------------------
diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/ElasticsearchGetRequest.java
b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/ElasticsearchGetRequest.java
new file mode 100644
index 0000000..442af3c
--- /dev/null
+++ b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/ElasticsearchGetRequest.java
@@ -0,0 +1,10 @@
+package org.apache.storm.elasticsearch;
+
+import org.elasticsearch.action.get.GetRequest;
+
+import backtype.storm.tuple.Tuple;
+
+public interface ElasticsearchGetRequest {
+
+    GetRequest extractFrom(Tuple tuple);
+}

http://git-wip-us.apache.org/repos/asf/storm/blob/59999924/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupBolt.java
----------------------------------------------------------------------
diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupBolt.java
b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupBolt.java
new file mode 100644
index 0000000..b6063d4
--- /dev/null
+++ b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupBolt.java
@@ -0,0 +1,70 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.storm.elasticsearch.bolt;
+
+import java.util.Collection;
+
+import org.apache.storm.elasticsearch.ElasticsearchGetRequest;
+import org.apache.storm.elasticsearch.common.EsConfig;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.GetResponse;
+
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+public class EsLookupBolt extends AbstractEsBolt {
+
+    private final ElasticsearchGetRequest getRequest;
+    private final EsLookupResultOutput output;
+
+    public EsLookupBolt(EsConfig esConfig, ElasticsearchGetRequest getRequest, EsLookupResultOutput
output) {
+        super(esConfig);
+        this.getRequest = getRequest;
+        this.output = output;
+    }
+
+    @Override
+    public void execute(Tuple tuple) {
+        try {
+            Collection<Values> values = lookupValuesInEs(tuple);
+            tryEmitAndAck(values, tuple);
+        } catch (Exception e) {
+            collector.reportError(e);
+            collector.fail(tuple);
+        }
+    }
+
+    private Collection<Values> lookupValuesInEs(Tuple tuple) {
+        GetRequest request = getRequest.extractFrom(tuple);
+        GetResponse response = client.get(request).actionGet();
+        return output.toValues(response);
+    }
+
+    private void tryEmitAndAck(Collection<Values> values, Tuple tuple) {
+        for (Values value : values) {
+            collector.emit(tuple, value);
+        }
+        collector.ack(tuple);
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
+        outputFieldsDeclarer.declare(output.fields());
+    }
+}

http://git-wip-us.apache.org/repos/asf/storm/blob/59999924/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupResultOutput.java
----------------------------------------------------------------------
diff --git a/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupResultOutput.java
b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupResultOutput.java
new file mode 100644
index 0000000..8696c6b
--- /dev/null
+++ b/external/storm-elasticsearch/src/main/java/org/apache/storm/elasticsearch/bolt/EsLookupResultOutput.java
@@ -0,0 +1,15 @@
+package org.apache.storm.elasticsearch.bolt;
+
+import java.util.Collection;
+
+import org.elasticsearch.action.get.GetResponse;
+
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Values;
+
+public interface EsLookupResultOutput {
+
+    Collection<Values> toValues(GetResponse response);
+
+    Fields fields();
+}

http://git-wip-us.apache.org/repos/asf/storm/blob/59999924/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltIntegrationTest.java
----------------------------------------------------------------------
diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltIntegrationTest.java
b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltIntegrationTest.java
new file mode 100644
index 0000000..2a544f9
--- /dev/null
+++ b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltIntegrationTest.java
@@ -0,0 +1,130 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.storm.elasticsearch.bolt;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.UUID;
+
+import org.apache.storm.elasticsearch.ElasticsearchGetRequest;
+import org.apache.storm.elasticsearch.common.EsConfig;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.GetResponse;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.verify;
+
+@RunWith(MockitoJUnitRunner.class)
+public class EsLookupBoltIntegrationTest extends AbstractEsBoltIntegrationTest<EsLookupBolt>
{
+
+    private final String documentId = UUID.randomUUID().toString();
+    private final String indexName = "index";
+    private final String typeName = "type";
+    private final String source = "{\"user\":\"user1\"}";
+
+    private ElasticsearchGetRequest getRequest = new TestElasticsearchGetRequest();
+    private EsLookupResultOutput output = new TestEsLookupResultOutput();
+
+    @Captor
+    private ArgumentCaptor<Tuple> anchor;
+
+    @Captor
+    private ArgumentCaptor<Values> emmitedValues;
+
+    @Mock
+    private Tuple tuple;
+
+    @Override
+    protected EsLookupBolt createBolt(EsConfig esConfig) {
+        return new EsLookupBolt(esConfig, getRequest, output);
+    }
+
+    @Before
+    public void populateIndexWithTestData() throws Exception {
+        node.client().prepareIndex(indexName, typeName, documentId).setSource(source).execute().actionGet();
+    }
+
+    @Before
+    public void clearIndex() throws Exception {
+        node.client().delete(new DeleteRequest(indexName, typeName, documentId)).actionGet();
+    }
+
+    @Test
+    public void anchorsTheTuple() throws Exception {
+        bolt.execute(tuple);
+
+        verify(outputCollector).emit(anchor.capture(), emmitedValues.capture());
+        assertThat(anchor.getValue(), is(tuple));
+    }
+
+    @Test
+    public void emitsExpectedValues() throws Exception {
+        Values expectedValues = expectedValues();
+
+        bolt.execute(tuple);
+
+        verify(outputCollector).emit(anchor.capture(), emmitedValues.capture());
+        assertThat(emmitedValues.getValue(), is(expectedValues));
+    }
+
+    @Test
+    public void acksTuple() throws Exception {
+        bolt.execute(tuple);
+
+        verify(outputCollector).ack(anchor.capture());
+        assertThat(anchor.getValue(), is(tuple));
+    }
+
+    private Values expectedValues() {
+        return new Values(source);
+    }
+
+    private class TestElasticsearchGetRequest implements ElasticsearchGetRequest {
+
+        @Override
+        public GetRequest extractFrom(Tuple tuple) {
+            return node.client().prepareGet().setId(documentId).setIndex(indexName).setType(typeName).request();
+        }
+    }
+
+    private class TestEsLookupResultOutput implements EsLookupResultOutput {
+
+        @Override
+        public Collection<Values> toValues(GetResponse response) {
+            return Collections.singleton(expectedValues());
+        }
+
+        @Override
+        public Fields fields() {
+            return new Fields("data");
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/storm/blob/59999924/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltTest.java
----------------------------------------------------------------------
diff --git a/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltTest.java
b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltTest.java
new file mode 100644
index 0000000..602ed6d
--- /dev/null
+++ b/external/storm-elasticsearch/src/test/java/org/apache/storm/elasticsearch/bolt/EsLookupBoltTest.java
@@ -0,0 +1,102 @@
+package org.apache.storm.elasticsearch.bolt;
+
+import java.util.Collections;
+import java.util.UUID;
+
+import org.apache.storm.elasticsearch.ElasticsearchGetRequest;
+import org.apache.storm.elasticsearch.common.EsConfig;
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.client.Client;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+
+import backtype.storm.topology.OutputFieldsDeclarer;
+import backtype.storm.tuple.Fields;
+import backtype.storm.tuple.Tuple;
+import backtype.storm.tuple.Values;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class EsLookupBoltTest extends AbstractEsBoltTest<EsLookupBolt> {
+
+    @Mock
+    private EsConfig esConfig;
+
+    @Mock
+    private ElasticsearchGetRequest getRequest;
+
+    @Mock
+    private EsLookupResultOutput output;
+
+    @Mock
+    private Tuple tuple;
+
+    @Mock
+    private GetRequest request;
+
+    @Mock
+    private Client client;
+
+    private Client originalClient;
+
+    @Override
+    protected EsLookupBolt createBolt(EsConfig esConfig) {
+        originalClient = EsLookupBolt.getClient();
+        EsLookupBolt.replaceClient(this.client);
+        return new EsLookupBolt(esConfig, getRequest, output);
+    }
+
+    @After
+    public void replaceClientWithOriginal() throws Exception {
+        EsLookupBolt.replaceClient(originalClient);
+    }
+
+    @Before
+    public void configureBoltDependencies() throws Exception {
+        when(getRequest.extractFrom(tuple)).thenReturn(request);
+        when(output.toValues(any(GetResponse.class))).thenReturn(Collections.singleton(new
Values("")));
+    }
+
+    @Test
+    public void failsTupleWhenClientThrows() throws Exception {
+        when(client.get(request)).thenThrow(ElasticsearchException.class);
+        bolt.execute(tuple);
+
+        verify(outputCollector).fail(tuple);
+    }
+
+    @Test
+    public void reportsExceptionWhenClientThrows() throws Exception {
+        ElasticsearchException elasticsearchException = new ElasticsearchException("dummy");
+        when(client.get(request)).thenThrow(elasticsearchException);
+        bolt.execute(tuple);
+
+        verify(outputCollector).reportError(elasticsearchException);
+    }
+
+    @Test
+    public void fieldsAreDeclaredThroughProvidedOutput() throws Exception {
+        Fields fields = new Fields(UUID.randomUUID().toString());
+        when(output.fields()).thenReturn(fields);
+        OutputFieldsDeclarer declarer = mock(OutputFieldsDeclarer.class);
+        bolt.declareOutputFields(declarer);
+
+        ArgumentCaptor<Fields> declaredFields = ArgumentCaptor.forClass(Fields.class);
+        verify(declarer).declare(declaredFields.capture());
+
+        assertThat(declaredFields.getValue(), is(fields));
+    }
+}


Mime
View raw message