You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2015/03/12 11:36:38 UTC
[1/6] incubator-streams-examples git commit: fix parent pom reference
simplify thread start
Repository: incubator-streams-examples
Updated Branches:
refs/heads/master 073f0f39c -> 5a5553475
fix parent pom reference
simplify thread start
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/commit/7288c390
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/tree/7288c390
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/diff/7288c390
Branch: refs/heads/master
Commit: 7288c390358c43032f0809e86ab02b1d6e981572
Parents: 5ac85ed
Author: sblackmon <sb...@apache.org>
Authored: Tue Feb 24 15:28:21 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Tue Feb 24 15:30:32 2015 -0600
----------------------------------------------------------------------
local/mongo-elasticsearch-sync/pom.xml | 3 ++-
.../example/elasticsearch/MongoElasticsearchSync.java | 14 ++++----------
2 files changed, 6 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/7288c390/local/mongo-elasticsearch-sync/pom.xml
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/pom.xml b/local/mongo-elasticsearch-sync/pom.xml
index 79470ee..4d271e0 100644
--- a/local/mongo-elasticsearch-sync/pom.xml
+++ b/local/mongo-elasticsearch-sync/pom.xml
@@ -22,8 +22,9 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.apache.streams</groupId>
- <artifactId>streams-examples</artifactId>
+ <artifactId>streams-examples-local</artifactId>
<version>0.2-incubating-SNAPSHOT</version>
+ <relativePath>..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/7288c390/local/mongo-elasticsearch-sync/src/main/java/org/apache/streams/example/elasticsearch/MongoElasticsearchSync.java
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/src/main/java/org/apache/streams/example/elasticsearch/MongoElasticsearchSync.java b/local/mongo-elasticsearch-sync/src/main/java/org/apache/streams/example/elasticsearch/MongoElasticsearchSync.java
index 568a1a9..fccbf47 100644
--- a/local/mongo-elasticsearch-sync/src/main/java/org/apache/streams/example/elasticsearch/MongoElasticsearchSync.java
+++ b/local/mongo-elasticsearch-sync/src/main/java/org/apache/streams/example/elasticsearch/MongoElasticsearchSync.java
@@ -45,19 +45,10 @@ public class MongoElasticsearchSync implements Runnable {
private final static Logger LOGGER = LoggerFactory.getLogger(MongoElasticsearchSync.class);
- protected static ListeningExecutorService executor = MoreExecutors.listeningDecorator(newFixedThreadPoolWithQueueSize(5, 20));
-
- private static ExecutorService newFixedThreadPoolWithQueueSize(int nThreads, int queueSize) {
- return new ThreadPoolExecutor(nThreads, nThreads,
- 5000L, TimeUnit.MILLISECONDS,
- new ArrayBlockingQueue<Runnable>(queueSize, true), new ThreadPoolExecutor.CallerRunsPolicy());
- }
-
MongoElasticsearchSyncConfiguration config;
public MongoElasticsearchSync() {
this(new ComponentConfigurator<MongoElasticsearchSyncConfiguration>(MongoElasticsearchSyncConfiguration.class).detectConfiguration(StreamsConfigurator.getConfig()));
-
}
public MongoElasticsearchSync(MongoElasticsearchSyncConfiguration config) {
@@ -68,7 +59,9 @@ public class MongoElasticsearchSync implements Runnable {
{
LOGGER.info(StreamsConfigurator.config.toString());
- executor.submit(new MongoElasticsearchSync());
+ MongoElasticsearchSync sync = new MongoElasticsearchSync();
+
+ new Thread(sync).start();
}
@@ -80,6 +73,7 @@ public class MongoElasticsearchSync implements Runnable {
ElasticsearchPersistWriter elasticsearchPersistWriter = new ElasticsearchPersistWriter(config.getDestination());
Map<String, Object> streamConfig = Maps.newHashMap();
+ streamConfig.put(LocalStreamBuilder.STREAM_IDENTIFIER_KEY, STREAMS_ID);
streamConfig.put(LocalStreamBuilder.TIMEOUT_KEY, 7 * 24 * 60 * 1000);
StreamBuilder builder = new LocalStreamBuilder(1000, streamConfig);
[5/6] incubator-streams-examples git commit: added docker
profile/plugin
Posted by sb...@apache.org.
added docker profile/plugin
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/commit/9d84e49c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/tree/9d84e49c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/diff/9d84e49c
Branch: refs/heads/master
Commit: 9d84e49cb649f62ada633a608d0fd0fd920f20c5
Parents: d7ec754
Author: sblackmon <sb...@apache.org>
Authored: Fri Feb 20 11:18:00 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Tue Feb 24 15:30:32 2015 -0600
----------------------------------------------------------------------
local/mongo-elasticsearch-sync/pom.xml | 40 +++++++++++++++++++++++++++++
1 file changed, 40 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/9d84e49c/local/mongo-elasticsearch-sync/pom.xml
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/pom.xml b/local/mongo-elasticsearch-sync/pom.xml
index 8e900a9..79470ee 100644
--- a/local/mongo-elasticsearch-sync/pom.xml
+++ b/local/mongo-elasticsearch-sync/pom.xml
@@ -310,4 +310,44 @@
</plugin>
</plugins>
</build>
+
+ <profiles>
+ <profile>
+ <id>docker</id>
+ <build>
+ <plugins>
+ <plugin>
+ <!-- The Docker Maven plugin is used to create docker image with the fat jar -->
+ <groupId>org.jolokia</groupId>
+ <artifactId>docker-maven-plugin</artifactId>
+ <version>0.11.0</version>
+ <configuration>
+ <images>
+
+ <image>
+ <alias>${project.artifactId}</alias>
+ <name>${project.artifactId}:${project.version}</name>
+ <build>
+ <from>dockerfile/java:oracle-java8</from>
+ <assembly>
+ <basedir>/</basedir>
+ <descriptorRef>artifact</descriptorRef>
+ </assembly>
+ <!-- Default command for the build image -->
+ </build>
+
+ </image>
+
+ </images>
+ </configuration>
+
+ </plugin>
+
+ </plugins>
+ </build>
+ <activation>
+ <activeByDefault/>
+ </activation>
+ </profile>
+ </profiles>
</project>
\ No newline at end of file
[3/6] incubator-streams-examples git commit: Example stream that
copies documents from mongodb to elasticsearch resolves STREAMS-289
Posted by sb...@apache.org.
Example stream that copies documents from mongodb to elasticsearch
resolves STREAMS-289
Note that for integration tests to pass, incubator-streams needs https://github.com/apache/incubator-streams/pull/191 to merge
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/commit/9e5cc48f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/tree/9e5cc48f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/diff/9e5cc48f
Branch: refs/heads/master
Commit: 9e5cc48f8d4397c4056d3f705e8c29761bc7081b
Parents: a05d4c9
Author: sblackmon <sb...@apache.org>
Authored: Wed Feb 18 18:13:04 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Tue Feb 24 15:30:32 2015 -0600
----------------------------------------------------------------------
.../MongoElasticsearchSync.png | Bin 0 -> 11502 bytes
local/mongo-elasticsearch-sync/README.md | 68 +++++
local/mongo-elasticsearch-sync/pom.xml | 269 +++++++++++++++++++
.../elasticsearch/MongoElasticsearchSync.java | 90 +++++++
.../MongoElasticsearchSyncConfiguration.json | 10 +
.../main/resources/MongoElasticsearchSync.dot | 18 ++
.../src/main/resources/application.json | 17 ++
.../test/MongoElasticsearchSyncIT.java | 112 ++++++++
.../src/test/resources/testSync.json | 18 ++
local/pom.xml | 1 +
10 files changed, 603 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/9e5cc48f/local/mongo-elasticsearch-sync/MongoElasticsearchSync.png
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/MongoElasticsearchSync.png b/local/mongo-elasticsearch-sync/MongoElasticsearchSync.png
new file mode 100644
index 0000000..a8fc4d7
Binary files /dev/null and b/local/mongo-elasticsearch-sync/MongoElasticsearchSync.png differ
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/9e5cc48f/local/mongo-elasticsearch-sync/README.md
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/README.md b/local/mongo-elasticsearch-sync/README.md
new file mode 100644
index 0000000..5822ae2
--- /dev/null
+++ b/local/mongo-elasticsearch-sync/README.md
@@ -0,0 +1,68 @@
+Apache Streams (incubating)
+Licensed under Apache License 2.0 - http://www.apache.org/licenses/LICENSE-2.0
+--------------------------------------------------------------------------------
+
+mongo-elasticsearch-sync
+==============================
+
+Requirements:
+-------------
+ - A running MongoDB 2.4+ instance
+ - A running ElasticSearch 1.0.0+ instance
+
+Description:
+------------
+Copies documents from mongodb to elasticsearch
+
+Specification:
+-----------------
+
+[MongoElasticsearchSync.dot](src/main/resources/MongoElasticsearchSync.dot "MongoElasticsearchSync.dot" )
+
+Diagram:
+-----------------
+
+
+
+Example Configuration:
+----------------------
+
+ {
+ "source": {
+ "host": "localhost",
+ "port": 27017,
+ "db": "streams",
+ "collection": "activities"
+ },
+ "destination": {
+ "hosts": [
+ "localhost"
+ ],
+ "port": 9300,
+ "clusterName": "elasticsearch",
+ "index": "destination",
+ "type": "activity"
+ }
+ }
+
+Build:
+---------
+
+`mvn clean package verify`
+
+Run:
+--------
+
+`java -cp target/mongo-elasticsearch-sync-0.1-SNAPSHOT.jar -Dconfig.file=src/main/resources/application.json org.apache.streams.example.elasticsearch.MongoElasticsearchSync`
+
+Deploy:
+--------
+
+`mvn -Pdocker clean package docker:build`
+
+`docker tag mongo-elasticsearch-sync:0.2-incubating-SNAPSHOT <dockerregistry>:mongo-elasticsearch-sync:0.2-incubating-SNAPSHOT`
+
+`docker push <dockerregistry>:mongo-elasticsearch-sync:0.2-incubating-SNAPSHOT`
+
+`docker run <dockerregistry>:mongo-elasticsearch-sync:0.2-incubating-SNAPSHOT.jar java -cp mongo-elasticsearch-sync-0.2-incubating-SNAPSHOT.jar -Dconfig.file=http://<location_of_config_file>.json org.apache.streams.example.elasticsearch.MongoElasticsearchSync`
+
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/9e5cc48f/local/mongo-elasticsearch-sync/pom.xml
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/pom.xml b/local/mongo-elasticsearch-sync/pom.xml
new file mode 100644
index 0000000..02a898b
--- /dev/null
+++ b/local/mongo-elasticsearch-sync/pom.xml
@@ -0,0 +1,269 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-examples</artifactId>
+ <version>0.2-incubating-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>mongo-elasticsearch-sync</artifactId>
+
+ <properties>
+ <elasticsearch.version>1.1.0</elasticsearch.version>
+ <lucene.version>4.7.2</lucene.version>
+ </properties>
+
+ <dependencies>
+ <!-- Test includes -->
+ <dependency>
+ <groupId>org.apache.lucene</groupId>
+ <artifactId>lucene-test-framework</artifactId>
+ <version>${lucene.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.lucene</groupId>
+ <artifactId>lucene-codecs</artifactId>
+ <version>${lucene.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.elasticsearch</groupId>
+ <artifactId>elasticsearch</artifactId>
+ <version>${elasticsearch.version}</version>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest-all</artifactId>
+ <version>1.3</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.typesafe</groupId>
+ <artifactId>config</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-config</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-runtime-local</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-persist-elasticsearch</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-persist-mongo</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-pojo</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ <version>${logback.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-core</artifactId>
+ <version>${logback.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <sourceDirectory>src/main/java</sourceDirectory>
+ <testSourceDirectory>src/test/java</testSourceDirectory>
+ <resources>
+ <resource>
+ <directory>src/main/resources</directory>
+ </resource>
+ </resources>
+ <testResources>
+ <testResource>
+ <directory>src/test/resources</directory>
+ </testResource>
+ </testResources>
+ <plugins>
+ <plugin>
+ <artifactId>maven-clean-plugin</artifactId>
+ <configuration>
+ <filesets>
+ <fileset>
+ <directory>data</directory>
+ <followSymlinks>false</followSymlinks>
+ </fileset>
+ </filesets>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>com.github.joelittlejohn.embedmongo</groupId>
+ <artifactId>embedmongo-maven-plugin</artifactId>
+ <version>0.1.12</version>
+ <executions>
+ <execution>
+ <id>start</id>
+ <goals>
+ <goal>start</goal>
+ </goals>
+ <configuration>
+ <port>37017</port>
+ <version>2.4.0</version>
+ <databaseDirectory>target/mongotest</databaseDirectory>
+ <logging>console</logging>
+ <skip>false</skip>
+ </configuration>
+ </execution>
+ <execution>
+ <id>stop</id>
+ <goals>
+ <goal>stop</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <finalName>${project.build.finalName}</finalName>
+ <filters>
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+ <exclude>META-INF/*.SF</exclude>
+ <exclude>META-INF/*.DSA</exclude>
+ <exclude>META-INF/*.RSA</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
+ <transformer
+ implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>org.apache.streams.moreover.example.MoreoverMetabaseHdfs</mainClass>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.jsonschema2pojo</groupId>
+ <artifactId>jsonschema2pojo-maven-plugin</artifactId>
+ <version>0.4.1</version>
+ <configuration>
+ <addCompileSourceRoot>true</addCompileSourceRoot>
+ <generateBuilders>true</generateBuilders>
+ <sourcePaths>
+ <sourcePath>src/main/jsonschema</sourcePath>
+ </sourcePaths>
+ <outputDirectory>target/generated-sources/jsonschema2pojo</outputDirectory>
+ <targetPackage>org.apache.streams.example.elasticsearch</targetPackage>
+ <useJodaDates>false</useJodaDates>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>generate</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>add-source</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <configuration>
+ <sources>
+ <source>target/generated-sources/jsonschema2pojo</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <version>2.4</version>
+ <executions>
+ <execution>
+ <id>resource-dependencies</id>
+ <phase>process-test-resources</phase>
+ <goals>
+ <goal>unpack-dependencies</goal>
+ </goals>
+ <configuration>
+ <includeArtifactIds>streams-pojo</includeArtifactIds>
+ <includes>**/*.json</includes>
+ <outputDirectory>${project.build.directory}/test-classes</outputDirectory>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <version>2.12.4</version>
+ <executions>
+ <execution>
+ <id>integration-tests</id>
+ <goals>
+ <goal>integration-test</goal>
+ <goal>verify</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/9e5cc48f/local/mongo-elasticsearch-sync/src/main/java/org/apache/streams/example/elasticsearch/MongoElasticsearchSync.java
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/src/main/java/org/apache/streams/example/elasticsearch/MongoElasticsearchSync.java b/local/mongo-elasticsearch-sync/src/main/java/org/apache/streams/example/elasticsearch/MongoElasticsearchSync.java
new file mode 100644
index 0000000..568a1a9
--- /dev/null
+++ b/local/mongo-elasticsearch-sync/src/main/java/org/apache/streams/example/elasticsearch/MongoElasticsearchSync.java
@@ -0,0 +1,90 @@
+/*
+ * 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
+ *
+ * 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.streams.example.elasticsearch;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.Maps;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.typesafe.config.Config;
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.elasticsearch.*;
+import org.apache.streams.core.StreamBuilder;
+import org.apache.streams.local.builders.LocalStreamBuilder;
+import org.apache.streams.mongo.MongoPersistReader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.*;
+
+/**
+ * Copies documents into a new index
+ */
+public class MongoElasticsearchSync implements Runnable {
+
+ public final static String STREAMS_ID = "MongoElasticsearchSync";
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(MongoElasticsearchSync.class);
+
+ protected static ListeningExecutorService executor = MoreExecutors.listeningDecorator(newFixedThreadPoolWithQueueSize(5, 20));
+
+ private static ExecutorService newFixedThreadPoolWithQueueSize(int nThreads, int queueSize) {
+ return new ThreadPoolExecutor(nThreads, nThreads,
+ 5000L, TimeUnit.MILLISECONDS,
+ new ArrayBlockingQueue<Runnable>(queueSize, true), new ThreadPoolExecutor.CallerRunsPolicy());
+ }
+
+ MongoElasticsearchSyncConfiguration config;
+
+ public MongoElasticsearchSync() {
+ this(new ComponentConfigurator<MongoElasticsearchSyncConfiguration>(MongoElasticsearchSyncConfiguration.class).detectConfiguration(StreamsConfigurator.getConfig()));
+
+ }
+
+ public MongoElasticsearchSync(MongoElasticsearchSyncConfiguration config) {
+ this.config = config;
+ }
+
+ public static void main(String[] args)
+ {
+ LOGGER.info(StreamsConfigurator.config.toString());
+
+ executor.submit(new MongoElasticsearchSync());
+
+ }
+
+ @Override
+ public void run() {
+
+ MongoPersistReader mongoPersistReader = new MongoPersistReader(config.getSource());
+
+ ElasticsearchPersistWriter elasticsearchPersistWriter = new ElasticsearchPersistWriter(config.getDestination());
+
+ Map<String, Object> streamConfig = Maps.newHashMap();
+ streamConfig.put(LocalStreamBuilder.TIMEOUT_KEY, 7 * 24 * 60 * 1000);
+ StreamBuilder builder = new LocalStreamBuilder(1000, streamConfig);
+
+ builder.newPerpetualStream(MongoPersistReader.STREAMS_ID, mongoPersistReader);
+ builder.addStreamsPersistWriter(ElasticsearchPersistWriter.STREAMS_ID, elasticsearchPersistWriter, 1, MongoPersistReader.STREAMS_ID);
+ builder.start();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/9e5cc48f/local/mongo-elasticsearch-sync/src/main/jsonschema/MongoElasticsearchSyncConfiguration.json
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/src/main/jsonschema/MongoElasticsearchSyncConfiguration.json b/local/mongo-elasticsearch-sync/src/main/jsonschema/MongoElasticsearchSyncConfiguration.json
new file mode 100644
index 0000000..a592699
--- /dev/null
+++ b/local/mongo-elasticsearch-sync/src/main/jsonschema/MongoElasticsearchSyncConfiguration.json
@@ -0,0 +1,10 @@
+{
+ "$schema": "http://json-schema.org/draft-03/schema",
+ "type": "object",
+ "javaType" : "org.apache.streams.example.elasticsearch.MongoElasticsearchSyncConfiguration",
+ "javaInterfaces": ["java.io.Serializable"],
+ "properties": {
+ "source": { "javaType": "org.apache.streams.mongo.MongoConfiguration", "type": "object", "required": true },
+ "destination": { "javaType": "org.apache.streams.elasticsearch.ElasticsearchWriterConfiguration", "type": "object", "required": true }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/9e5cc48f/local/mongo-elasticsearch-sync/src/main/resources/MongoElasticsearchSync.dot
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/src/main/resources/MongoElasticsearchSync.dot b/local/mongo-elasticsearch-sync/src/main/resources/MongoElasticsearchSync.dot
new file mode 100644
index 0000000..e3babf5
--- /dev/null
+++ b/local/mongo-elasticsearch-sync/src/main/resources/MongoElasticsearchSync.dot
@@ -0,0 +1,18 @@
+digraph g {
+
+ //providers
+ MongoPersistReader [label="MongoPersistReader",shape=ellipse,URL="https://github.com/apache/incubator-streams/blob/master/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistReader.java"];
+
+ //persisters
+ ElasticsearchPersistWriter [label="ElasticsearchPersistWriter",shape=ellipse,URL="https://github.com/apache/incubator-streams/blob/master/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java"];
+
+ //data
+ source [label="mongdb://{db}/{collection}",shape=box];
+ destination [label="es://{index}/{type}",shape=box];
+
+ //stream
+ source -> MongoPersistReader [label="ObjectNode"];
+ MongoPersistReader -> ElasticsearchPersistWriter [label="ObjectNode"];
+ ElasticsearchPersistWriter -> destination [label="ObjectNode"];
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/9e5cc48f/local/mongo-elasticsearch-sync/src/main/resources/application.json
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/src/main/resources/application.json b/local/mongo-elasticsearch-sync/src/main/resources/application.json
new file mode 100644
index 0000000..facb336
--- /dev/null
+++ b/local/mongo-elasticsearch-sync/src/main/resources/application.json
@@ -0,0 +1,17 @@
+{
+ "source": {
+ "host": "localhost",
+ "port": 27017,
+ "db": "streams",
+ "collection": "activities"
+ },
+ "destination": {
+ "hosts": [
+ "localhost"
+ ],
+ "port": 9300,
+ "clusterName": "elasticsearch",
+ "index": "destination",
+ "type": "activity"
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/9e5cc48f/local/mongo-elasticsearch-sync/src/test/java/org/apache/streams/example/elasticsearch/test/MongoElasticsearchSyncIT.java
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/src/test/java/org/apache/streams/example/elasticsearch/test/MongoElasticsearchSyncIT.java b/local/mongo-elasticsearch-sync/src/test/java/org/apache/streams/example/elasticsearch/test/MongoElasticsearchSyncIT.java
new file mode 100644
index 0000000..df6fa00
--- /dev/null
+++ b/local/mongo-elasticsearch-sync/src/test/java/org/apache/streams/example/elasticsearch/test/MongoElasticsearchSyncIT.java
@@ -0,0 +1,112 @@
+/*
+ * 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
+ *
+ * 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.streams.example.elasticsearch.test;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.Lists;
+import org.apache.commons.io.Charsets;
+import org.apache.commons.io.IOUtils;
+import org.apache.streams.config.ComponentConfigurator;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.elasticsearch.ElasticsearchConfiguration;
+import org.apache.streams.elasticsearch.ElasticsearchPersistWriter;
+import org.apache.streams.elasticsearch.ElasticsearchWriterConfiguration;
+import org.apache.streams.example.elasticsearch.MongoElasticsearchSync;
+import org.apache.streams.example.elasticsearch.MongoElasticsearchSyncConfiguration;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.mongo.MongoConfiguration;
+import org.apache.streams.mongo.MongoPersistReader;
+import org.apache.streams.mongo.MongoPersistWriter;
+import org.apache.streams.pojo.json.Activity;
+import org.elasticsearch.test.ElasticsearchIntegrationTest;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.InputStream;
+import java.util.List;
+
+/**
+ * Test copying documents between two indexes on same cluster
+ */
+@ElasticsearchIntegrationTest.ClusterScope(scope= ElasticsearchIntegrationTest.Scope.TEST, numNodes=1)
+public class MongoElasticsearchSyncIT extends ElasticsearchIntegrationTest {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(MongoElasticsearchSyncIT.class);
+
+ ObjectMapper MAPPER = StreamsJacksonMapper.getInstance();
+
+ MongoElasticsearchSyncConfiguration syncConfiguration;
+
+ int srcCount = 0;
+
+ @Before
+ public void prepareTest() throws Exception {
+
+ syncConfiguration = MAPPER.readValue(
+ MongoElasticsearchSyncIT.class.getResourceAsStream("/testSync.json"), MongoElasticsearchSyncConfiguration.class);
+
+ syncConfiguration.getDestination().setClusterName(cluster().getClusterName());
+
+ MongoPersistWriter setupWriter = new MongoPersistWriter(syncConfiguration.getSource());
+
+ setupWriter.prepare(null);
+
+ InputStream testActivityFolderStream = MongoElasticsearchSyncIT.class.getClassLoader()
+ .getResourceAsStream("activities");
+ List<String> files = IOUtils.readLines(testActivityFolderStream, Charsets.UTF_8);
+
+ for( String file : files) {
+ LOGGER.info("File: " + file );
+ InputStream testActivityFileStream = MongoElasticsearchSyncIT.class.getClassLoader()
+ .getResourceAsStream("activities/" + file);
+ Activity activity = MAPPER.readValue(testActivityFileStream, Activity.class);
+ activity.getAdditionalProperties().remove("$license");
+ StreamsDatum datum = new StreamsDatum(activity, activity.getVerb());
+ setupWriter.write( datum );
+ LOGGER.info("Wrote: " + activity.getVerb() );
+ srcCount++;
+ }
+
+ setupWriter.cleanUp();
+
+ }
+
+ @Test
+ public void testSync() throws Exception {
+
+ assert srcCount > 0;
+
+ MongoElasticsearchSync sync = new MongoElasticsearchSync(syncConfiguration);
+
+ Thread reindexThread = new Thread(sync);
+ reindexThread.start();
+ reindexThread.join();
+
+ flushAndRefresh();
+
+ assert(indexExists("destination"));
+
+ long destCount = client().count(client().prepareCount("destination").request()).get().getCount();
+ assert srcCount == destCount;
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/9e5cc48f/local/mongo-elasticsearch-sync/src/test/resources/testSync.json
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/src/test/resources/testSync.json b/local/mongo-elasticsearch-sync/src/test/resources/testSync.json
new file mode 100644
index 0000000..b678c21
--- /dev/null
+++ b/local/mongo-elasticsearch-sync/src/test/resources/testSync.json
@@ -0,0 +1,18 @@
+{
+ "source": {
+ "host": "localhost",
+ "port": 37017,
+ "db": "local",
+ "collection": "activities"
+ },
+ "destination": {
+ "hosts": [
+ "localhost"
+ ],
+ "port": 9300,
+ "clusterName": "elasticsearch",
+ "index": "destination",
+ "type": "activity",
+ "forceUseConfig": true
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/9e5cc48f/local/pom.xml
----------------------------------------------------------------------
diff --git a/local/pom.xml b/local/pom.xml
index 184d463..e58c23f 100644
--- a/local/pom.xml
+++ b/local/pom.xml
@@ -38,6 +38,7 @@
<modules>
<module>elasticsearch-reindex</module>
+ <module>mongo-elasticsearch-sync</module>
</modules>
</project>
[6/6] incubator-streams-examples git commit: Merge branch
'STREAMS-289'
Posted by sb...@apache.org.
Merge branch 'STREAMS-289'
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/commit/5a555347
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/tree/5a555347
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/diff/5a555347
Branch: refs/heads/master
Commit: 5a5553475adf23b306161ee8430ddcf706a59c2a
Parents: 073f0f3 7288c39
Author: sblackmon <sb...@apache.org>
Authored: Thu Mar 12 03:36:22 2015 -0700
Committer: sblackmon <sb...@apache.org>
Committed: Thu Mar 12 03:36:22 2015 -0700
----------------------------------------------------------------------
.../MongoElasticsearchSync.png | Bin 0 -> 11502 bytes
local/mongo-elasticsearch-sync/README.md | 68 ++++
local/mongo-elasticsearch-sync/pom.xml | 354 +++++++++++++++++++
.../elasticsearch/MongoElasticsearchSync.java | 84 +++++
.../MongoElasticsearchSyncConfiguration.json | 10 +
.../main/resources/MongoElasticsearchSync.dot | 18 +
.../src/main/resources/application.json | 17 +
.../test/MongoElasticsearchSyncIT.java | 112 ++++++
.../src/test/resources/testSync.json | 18 +
local/pom.xml | 1 +
10 files changed, 682 insertions(+)
----------------------------------------------------------------------
[2/6] incubator-streams-examples git commit: docker details in README
Posted by sb...@apache.org.
docker details in README
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/commit/5ac85ed6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/tree/5ac85ed6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/diff/5ac85ed6
Branch: refs/heads/master
Commit: 5ac85ed68432869338ebb6de8a31e193dc96ac75
Parents: 9d84e49
Author: sblackmon <sb...@apache.org>
Authored: Sat Feb 21 14:02:48 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Tue Feb 24 15:30:32 2015 -0600
----------------------------------------------------------------------
local/mongo-elasticsearch-sync/README.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/5ac85ed6/local/mongo-elasticsearch-sync/README.md
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/README.md b/local/mongo-elasticsearch-sync/README.md
index 5822ae2..72b11db 100644
--- a/local/mongo-elasticsearch-sync/README.md
+++ b/local/mongo-elasticsearch-sync/README.md
@@ -64,5 +64,5 @@ Deploy:
`docker push <dockerregistry>:mongo-elasticsearch-sync:0.2-incubating-SNAPSHOT`
-`docker run <dockerregistry>:mongo-elasticsearch-sync:0.2-incubating-SNAPSHOT.jar java -cp mongo-elasticsearch-sync-0.2-incubating-SNAPSHOT.jar -Dconfig.file=http://<location_of_config_file>.json org.apache.streams.example.elasticsearch.MongoElasticsearchSync`
+`docker run <dockerregistry>:mongo-elasticsearch-sync:0.2-incubating-SNAPSHOT java -cp mongo-elasticsearch-sync-0.2-incubating-SNAPSHOT.jar -Dconfig.file=http://<location_of_config_file>.json org.apache.streams.example.elasticsearch.MongoElasticsearchSync`
[4/6] incubator-streams-examples git commit: change mainClass logging
dependencies & enforcer
Posted by sb...@apache.org.
change mainClass
logging dependencies & enforcer
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/commit/d7ec754a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/tree/d7ec754a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/diff/d7ec754a
Branch: refs/heads/master
Commit: d7ec754a032189e467ad8b60a3672c536bb6230a
Parents: 9e5cc48
Author: sblackmon <sb...@apache.org>
Authored: Thu Feb 19 11:08:35 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Tue Feb 24 15:30:32 2015 -0600
----------------------------------------------------------------------
local/mongo-elasticsearch-sync/pom.xml | 46 ++++++++++++++++++++++++++++-
1 file changed, 45 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/d7ec754a/local/mongo-elasticsearch-sync/pom.xml
----------------------------------------------------------------------
diff --git a/local/mongo-elasticsearch-sync/pom.xml b/local/mongo-elasticsearch-sync/pom.xml
index 02a898b..8e900a9 100644
--- a/local/mongo-elasticsearch-sync/pom.xml
+++ b/local/mongo-elasticsearch-sync/pom.xml
@@ -96,6 +96,21 @@
<type>test-jar</type>
</dependency>
<dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <version>${slf4j.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <version>${slf4j.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jul-to-slf4j</artifactId>
+ <version>${slf4j.version}</version>
+ </dependency>
+ <dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>${logback.version}</version>
@@ -121,6 +136,35 @@
</testResource>
</testResources>
<plugins>
+ <!-- This binary runs with logback -->
+ <!-- Keep log4j out -->
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-enforcer-plugin</artifactId>
+ <version>1.3.1</version>
+ <executions>
+ <execution>
+ <id>enforce-banned-dependencies</id>
+ <goals>
+ <goal>enforce</goal>
+ </goals>
+ <configuration>
+ <rules>
+ <bannedDependencies>
+ <excludes>
+ <exclude>org.slf4j:slf4j-log4j12</exclude>
+ <exclude>org.slf4j:slf4j-jcl</exclude>
+ <exclude>org.slf4j:slf4j-jdk14</exclude>
+ <exclude>org.log4j:log4j</exclude>
+ <exclude>commons-logging:commons-logging</exclude>
+ </excludes>
+ </bannedDependencies>
+ </rules>
+ <fail>true</fail>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
<plugin>
<artifactId>maven-clean-plugin</artifactId>
<configuration>
@@ -184,7 +228,7 @@
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
- <mainClass>org.apache.streams.moreover.example.MoreoverMetabaseHdfs</mainClass>
+ <mainClass>org.apache.streams.example.elasticsearch.MongoElasticsearchSync</mainClass>
</transformer>
</transformers>
</configuration>