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/02/20 18:21:24 UTC
[1/2] incubator-streams-examples git commit: Example stream that
copies documents into a new index.
Repository: incubator-streams-examples
Updated Branches:
refs/heads/master 076b3ac16 -> a05d4c9bf
Example stream that copies documents into a new index.
For integration tests to work, the following need to merge into incubator-streams:
STREAMS-285
STREAMS-286
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/76c26e76
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/tree/76c26e76
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/diff/76c26e76
Branch: refs/heads/master
Commit: 76c26e76c951b0be60a92ed83465fc7cdb57c290
Parents: 076b3ac
Author: sblackmon <sb...@apache.org>
Authored: Wed Feb 18 12:07:59 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Wed Feb 18 12:07:59 2015 -0600
----------------------------------------------------------------------
.../ElasticsearchReindex.png | Bin 0 -> 8913 bytes
local/elasticsearch-reindex/README.md | 79 +++++
local/elasticsearch-reindex/pom.xml | 313 +++++++++++++++++++
.../example/ElasticsearchReindex.java | 89 ++++++
.../ElasticsearchReindexConfiguration.json | 10 +
.../src/main/resources/ElasticsearchReindex.dot | 17 +
.../src/main/resources/application.json | 26 ++
.../src/main/resources/log4j.properties | 8 +
.../src/main/resources/logback.xml | 27 ++
.../test/ElasticsearchReindexIT.java | 116 +++++++
.../test/ElasticsearchReindexParentChildIT.java | 162 ++++++++++
.../resources/ActivityChildObjectParent.json | 15 +
.../src/test/resources/testReindex.json | 26 ++
.../src/test/resources/testReindexChild.json | 26 ++
.../src/test/resources/testReindexParent.json | 26 ++
local/pom.xml | 1 +
16 files changed, 941 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/76c26e76/local/elasticsearch-reindex/ElasticsearchReindex.png
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/ElasticsearchReindex.png b/local/elasticsearch-reindex/ElasticsearchReindex.png
new file mode 100644
index 0000000..9e59093
Binary files /dev/null and b/local/elasticsearch-reindex/ElasticsearchReindex.png differ
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/76c26e76/local/elasticsearch-reindex/README.md
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/README.md b/local/elasticsearch-reindex/README.md
new file mode 100644
index 0000000..5fbace2
--- /dev/null
+++ b/local/elasticsearch-reindex/README.md
@@ -0,0 +1,79 @@
+Apache Streams (incubating)
+Licensed under Apache License 2.0 - http://www.apache.org/licenses/LICENSE-2.0
+--------------------------------------------------------------------------------
+
+elasticsearch-reindex
+==============================
+
+Requirements:
+-------------
+ - A running ElasticSearch 1.0.0+ cluster
+ - Transport client access to cluster
+ - elasticsearch.version and lucene.version set to match cluster
+
+Description:
+------------
+Copies documents into a different index
+
+Specification:
+-----------------
+
+[ElasticsearchReindex.dot](src/main/resources/ElasticsearchReindex.dot "ElasticsearchReindex.dot" )
+
+Diagram:
+-----------------
+
+
+
+Example Configuration:
+----------------------
+
+ {
+ "source": {
+ "hosts": [
+ "localhost"
+ ],
+ "port": 9300,
+ "clusterName": "elasticsearch",
+ "indexes": [
+ "activity"
+ ],
+ "types": [
+ "activity"
+ ],
+ "forceUseConfig": true
+ },
+ "destination": {
+ "hosts": [
+ "localhost"
+ ],
+ "port": 9300,
+ "clusterName": "elasticsearch",
+ "index": "activity2",
+ "type": "activity",
+ "forceUseConfig": true
+ }
+ }
+
+Populate source and destination in configuration with cluster / index / type details.
+
+Build:
+---------
+
+`mvn clean package verify`
+
+Run:
+--------
+
+`java -cp target/elasticsearch-reindex-0.1-SNAPSHOT.jar -Dconfig.file=src/main/resources/application.json org.apache.streams.example.elasticsearch.ElasticsearchReindex`
+
+Deploy:
+--------
+`mvn -Pdocker clean package docker:build`
+
+`docker tag elasticsearch-reindex:0.2-incubating-SNAPSHOT <dockerregistry>:elasticsearch-reindex:0.2-incubating-SNAPSHOT`
+
+`docker push <dockerregistry>:elasticsearch-reindex:0.2-incubating-SNAPSHOT`
+
+`docker run <dockerregistry>:elasticsearch-reindex:0.2-incubating-SNAPSHOT.jar java -cp elasticsearch-reindex-0.2-incubating-SNAPSHOT.jar -Dconfig.file=http://<location_of_config_file>.json org.apache.streams.example.elasticsearch.ElasticsearchReindex`
+
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/76c26e76/local/elasticsearch-reindex/pom.xml
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/pom.xml b/local/elasticsearch-reindex/pom.xml
new file mode 100644
index 0000000..4bb6be3
--- /dev/null
+++ b/local/elasticsearch-reindex/pom.xml
@@ -0,0 +1,313 @@
+<?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-local</artifactId>
+ <version>0.2-incubating-SNAPSHOT</version>
+ <relativePath>..</relativePath>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>elasticsearch-reindex</artifactId>
+ <version>0.2-incubating-SNAPSHOT</version>
+
+ <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-util</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-pojo</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ </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>0.2-incubating-SNAPSHOT</version>
+ </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>
+ <!-- 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>
+ </excludes>
+ </bannedDependencies>
+ </rules>
+ <fail>true</fail>
+ </configuration>
+ </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>
+ <exclude>**/logback.xml</exclude>
+ <exclude>**/log4j.properties</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.example.elasticsearch.ElasticsearchReindex</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>
+
+ <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>
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/76c26e76/local/elasticsearch-reindex/src/main/java/org/apache/streams/elasticsearch/example/ElasticsearchReindex.java
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/main/java/org/apache/streams/elasticsearch/example/ElasticsearchReindex.java b/local/elasticsearch-reindex/src/main/java/org/apache/streams/elasticsearch/example/ElasticsearchReindex.java
new file mode 100644
index 0000000..37e178c
--- /dev/null
+++ b/local/elasticsearch-reindex/src/main/java/org/apache/streams/elasticsearch/example/ElasticsearchReindex.java
@@ -0,0 +1,89 @@
+/*
+ * 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.elasticsearch.example;
+
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.concurrent.*;
+
+/**
+ * Copies documents into a new index
+ */
+public class ElasticsearchReindex implements Runnable {
+
+ public final static String STREAMS_ID = "ElasticsearchReindex";
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(ElasticsearchReindex.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());
+ }
+
+ ElasticsearchReindexConfiguration reindex;
+
+ public ElasticsearchReindex() {
+ this(new ComponentConfigurator<ElasticsearchReindexConfiguration>(ElasticsearchReindexConfiguration.class).detectConfiguration(StreamsConfigurator.getConfig()));
+
+ }
+
+ public ElasticsearchReindex(ElasticsearchReindexConfiguration reindex) {
+ this.reindex = reindex;
+ }
+
+ public static void main(String[] args)
+ {
+ LOGGER.info(StreamsConfigurator.config.toString());
+
+ executor.submit(new ElasticsearchReindex());
+
+ }
+
+ @Override
+ public void run() {
+
+ ElasticsearchPersistReader elasticsearchPersistReader = new ElasticsearchPersistReader(reindex.getSource());
+
+ ElasticsearchPersistWriter elasticsearchPersistWriter = new ElasticsearchPersistWriter(reindex.getDestination());
+
+ Map<String, Object> streamConfig = Maps.newHashMap();
+ streamConfig.put(LocalStreamBuilder.TIMEOUT_KEY, 7 * 24 * 60 * 1000);
+ StreamBuilder builder = new LocalStreamBuilder(1000, streamConfig);
+
+ builder.newPerpetualStream(ElasticsearchPersistReader.STREAMS_ID, elasticsearchPersistReader);
+ builder.addStreamsPersistWriter(ElasticsearchPersistWriter.STREAMS_ID, elasticsearchPersistWriter, 1, ElasticsearchPersistReader.STREAMS_ID);
+ builder.start();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/76c26e76/local/elasticsearch-reindex/src/main/jsonschema/ElasticsearchReindexConfiguration.json
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/main/jsonschema/ElasticsearchReindexConfiguration.json b/local/elasticsearch-reindex/src/main/jsonschema/ElasticsearchReindexConfiguration.json
new file mode 100644
index 0000000..ffdc4d8
--- /dev/null
+++ b/local/elasticsearch-reindex/src/main/jsonschema/ElasticsearchReindexConfiguration.json
@@ -0,0 +1,10 @@
+{
+ "$schema": "http://json-schema.org/draft-03/schema",
+ "type": "object",
+ "javaType" : "org.apache.streams.elasticsearch.example.ElasticsearchReindexConfiguration",
+ "javaInterfaces": ["java.io.Serializable"],
+ "properties": {
+ "source": { "javaType": "org.apache.streams.elasticsearch.ElasticsearchReaderConfiguration", "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/76c26e76/local/elasticsearch-reindex/src/main/resources/ElasticsearchReindex.dot
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/main/resources/ElasticsearchReindex.dot b/local/elasticsearch-reindex/src/main/resources/ElasticsearchReindex.dot
new file mode 100644
index 0000000..52456a9
--- /dev/null
+++ b/local/elasticsearch-reindex/src/main/resources/ElasticsearchReindex.dot
@@ -0,0 +1,17 @@
+digraph g {
+
+ //providers
+ ElasticsearchPersistReader [label="ElasticsearchPersistReader",shape=ellipse,URL="https://github.com/apache/incubator-streams/blob/master/streams-contrib/streams-provider-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistReader.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="es://{indexes}/{types}",shape=box];
+ destination [label="es://{index}/{type}",shape=box];
+
+ //stream
+ source -> ElasticsearchPersistReader
+ ElasticsearchPersistReader -> ElasticsearchPersistWriter [label="String"];
+ ElasticsearchPersistWriter -> destination
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/76c26e76/local/elasticsearch-reindex/src/main/resources/application.json
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/main/resources/application.json b/local/elasticsearch-reindex/src/main/resources/application.json
new file mode 100644
index 0000000..567b19d
--- /dev/null
+++ b/local/elasticsearch-reindex/src/main/resources/application.json
@@ -0,0 +1,26 @@
+{
+ "source": {
+ "hosts": [
+ "localhost"
+ ],
+ "port": 9300,
+ "clusterName": "elasticsearch",
+ "indexes": [
+ "activity"
+ ],
+ "types": [
+ "activity"
+ ],
+ "forceUseConfig": true
+ },
+ "destination": {
+ "hosts": [
+ "localhost"
+ ],
+ "port": 9300,
+ "clusterName": "elasticsearch",
+ "index": "activity2",
+ "type": "activity",
+ "forceUseConfig": true
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/76c26e76/local/elasticsearch-reindex/src/main/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/main/resources/log4j.properties b/local/elasticsearch-reindex/src/main/resources/log4j.properties
new file mode 100644
index 0000000..71255bb
--- /dev/null
+++ b/local/elasticsearch-reindex/src/main/resources/log4j.properties
@@ -0,0 +1,8 @@
+# Root logger option
+log4j.rootLogger=DEBUG, stdout
+
+# Direct log messages to stdout
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.Target=System.out
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/76c26e76/local/elasticsearch-reindex/src/main/resources/logback.xml
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/main/resources/logback.xml b/local/elasticsearch-reindex/src/main/resources/logback.xml
new file mode 100644
index 0000000..fc47894
--- /dev/null
+++ b/local/elasticsearch-reindex/src/main/resources/logback.xml
@@ -0,0 +1,27 @@
+<configuration debug="true" scan="true" scanPeriod="5 seconds">
+
+ <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+ <append>true</append>
+ <!-- encoders are assigned the type
+ ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+ <encoder>
+ <pattern>%-4relative [%thread] %-5level %logger{35} - %msg%n</pattern>
+ </encoder>
+ </appender>
+
+ <appender name="FILE" class="ch.qos.logback.core.FileAppender">
+ <file>target/logback.txt</file>
+ <append>true</append>
+ <!-- encoders are assigned the type
+ ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+ <encoder>
+ <pattern>%-4relative [%thread] %-5level %logger{35} - %msg%n</pattern>
+ </encoder>
+ </appender>
+
+ <root level="DEBUG">
+ <appender-ref ref="CONSOLE" />
+ <appender-ref ref="FILE" />
+ </root>
+
+</configuration>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/76c26e76/local/elasticsearch-reindex/src/test/java/org/apache/streams/example/elasticsearch/test/ElasticsearchReindexIT.java
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/test/java/org/apache/streams/example/elasticsearch/test/ElasticsearchReindexIT.java b/local/elasticsearch-reindex/src/test/java/org/apache/streams/example/elasticsearch/test/ElasticsearchReindexIT.java
new file mode 100644
index 0000000..ebbe00e
--- /dev/null
+++ b/local/elasticsearch-reindex/src/test/java/org/apache/streams/example/elasticsearch/test/ElasticsearchReindexIT.java
@@ -0,0 +1,116 @@
+/*
+ * 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.core.StreamsDatum;
+import org.apache.streams.elasticsearch.ElasticsearchConfiguration;
+import org.apache.streams.elasticsearch.ElasticsearchPersistWriter;
+import org.apache.streams.elasticsearch.ElasticsearchWriterConfiguration;
+import org.apache.streams.elasticsearch.example.ElasticsearchReindex;
+import org.apache.streams.elasticsearch.example.ElasticsearchReindexConfiguration;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+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.*;
+
+/**
+ * Test copying documents between two indexes on same cluster
+ */
+@ElasticsearchIntegrationTest.ClusterScope(scope= ElasticsearchIntegrationTest.Scope.TEST, numNodes=1)
+public class ElasticsearchReindexIT extends ElasticsearchIntegrationTest {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(ElasticsearchReindexIT.class);
+
+ ObjectMapper MAPPER = StreamsJacksonMapper.getInstance();
+
+ ElasticsearchConfiguration testConfiguration = new ElasticsearchConfiguration();
+
+ @Before
+ public void prepareTest() throws Exception {
+
+ testConfiguration = new ElasticsearchConfiguration();
+ testConfiguration.setHosts(Lists.newArrayList("localhost"));
+ testConfiguration.setClusterName(cluster().getClusterName());
+
+ ElasticsearchWriterConfiguration setupWriterConfiguration = MAPPER.convertValue(testConfiguration, ElasticsearchWriterConfiguration.class);
+ setupWriterConfiguration.setIndex("source");
+ setupWriterConfiguration.setType("activity");
+ setupWriterConfiguration.setBatchSize(5l);
+
+ ElasticsearchPersistWriter setupWriter = new ElasticsearchPersistWriter(setupWriterConfiguration);
+ setupWriter.prepare(null);
+
+ InputStream testActivityFolderStream = ElasticsearchReindexIT.class.getClassLoader()
+ .getResourceAsStream("activities");
+ List<String> files = IOUtils.readLines(testActivityFolderStream, Charsets.UTF_8);
+
+ for( String file : files) {
+ LOGGER.info("File: " + file );
+ InputStream testActivityFileStream = ElasticsearchReindexIT.class.getClassLoader()
+ .getResourceAsStream("activities/" + file);
+ Activity activity = MAPPER.readValue(testActivityFileStream, Activity.class);
+ StreamsDatum datum = new StreamsDatum(activity, activity.getVerb());
+ setupWriter.write( datum );
+ LOGGER.info("Wrote: " + activity.getVerb() );
+ }
+
+ setupWriter.cleanUp();
+
+ flushAndRefresh();
+
+ }
+
+ @Test
+ public void testReindex() throws Exception {
+
+ ElasticsearchReindexConfiguration reindexConfiguration = MAPPER.readValue(
+ ElasticsearchReindexIT.class.getResourceAsStream("/testReindex.json"), ElasticsearchReindexConfiguration.class);
+
+ reindexConfiguration.getDestination().setClusterName(cluster().getClusterName());
+ reindexConfiguration.getSource().setClusterName(cluster().getClusterName());
+
+ assert(indexExists("source"));
+ long srcCount = client().count(client().prepareCount("source").request()).get().getCount();
+ assert srcCount > 0;
+
+ ElasticsearchReindex reindex = new ElasticsearchReindex(reindexConfiguration);
+
+ Thread reindexThread = new Thread(reindex);
+ 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/76c26e76/local/elasticsearch-reindex/src/test/java/org/apache/streams/example/elasticsearch/test/ElasticsearchReindexParentChildIT.java
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/test/java/org/apache/streams/example/elasticsearch/test/ElasticsearchReindexParentChildIT.java b/local/elasticsearch-reindex/src/test/java/org/apache/streams/example/elasticsearch/test/ElasticsearchReindexParentChildIT.java
new file mode 100644
index 0000000..95a7684
--- /dev/null
+++ b/local/elasticsearch-reindex/src/test/java/org/apache/streams/example/elasticsearch/test/ElasticsearchReindexParentChildIT.java
@@ -0,0 +1,162 @@
+/*
+ * 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.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Lists;
+import org.apache.commons.io.Charsets;
+import org.apache.commons.io.IOUtils;
+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.elasticsearch.example.ElasticsearchReindex;
+import org.apache.streams.elasticsearch.example.ElasticsearchReindexConfiguration;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+import org.elasticsearch.action.admin.indices.template.put.PutIndexTemplateRequestBuilder;
+import org.elasticsearch.test.ElasticsearchIntegrationTest;
+import org.junit.Before;
+import org.junit.Test;
+import org.reflections.Reflections;
+import org.reflections.scanners.SubTypesScanner;
+import org.reflections.util.ClasspathHelper;
+import org.reflections.util.ConfigurationBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.InputStream;
+import java.net.URL;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Test copying parent/child associated documents between two indexes on same cluster
+ */
+@ElasticsearchIntegrationTest.ClusterScope(scope= ElasticsearchIntegrationTest.Scope.TEST, numNodes=1)
+public class ElasticsearchReindexParentChildIT extends ElasticsearchIntegrationTest {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(ElasticsearchReindexIT.class);
+
+ ObjectMapper MAPPER = StreamsJacksonMapper.getInstance();
+
+ ElasticsearchConfiguration testConfiguration = new ElasticsearchConfiguration();
+
+ @Before
+ public void prepareTest() throws Exception {
+
+ testConfiguration = new ElasticsearchConfiguration();
+ testConfiguration.setHosts(Lists.newArrayList("localhost"));
+ testConfiguration.setClusterName(cluster().getClusterName());
+
+ PutIndexTemplateRequestBuilder putTemplateRequestBuilder = client().admin().indices().preparePutTemplate("mappings");
+ URL templateURL = ElasticsearchReindexParentChildIT.class.getResource("/ActivityChildObjectParent.json");
+ ObjectNode template = MAPPER.readValue(templateURL, ObjectNode.class);
+ String templateSource = MAPPER.writeValueAsString(template);
+ putTemplateRequestBuilder.setSource(templateSource);
+
+ client().admin().indices().putTemplate(putTemplateRequestBuilder.request()).actionGet();
+
+ Reflections reflections = new Reflections(new ConfigurationBuilder()
+ .setUrls(ClasspathHelper.forPackage("org.apache.streams.pojo.json"))
+ .setScanners(new SubTypesScanner()));
+ Set<Class<? extends ActivityObject>> objectTypes = reflections.getSubTypesOf(ActivityObject.class);
+
+ ElasticsearchWriterConfiguration setupWriterConfiguration = MAPPER.convertValue(testConfiguration, ElasticsearchWriterConfiguration.class);
+ setupWriterConfiguration.setIndex("source");
+ setupWriterConfiguration.setBatchSize(5l);
+
+ ElasticsearchPersistWriter setupWriter = new ElasticsearchPersistWriter(setupWriterConfiguration);
+ setupWriter.prepare(null);
+
+ for( Class objectType : objectTypes ) {
+ Object object = objectType.newInstance();
+ ActivityObject activityObject = MAPPER.convertValue(object, ActivityObject.class);
+ StreamsDatum datum = new StreamsDatum(activityObject, activityObject.getObjectType());
+ datum.getMetadata().put("type", "object");
+ setupWriter.write( datum );
+ }
+
+ InputStream testActivityFolderStream = ElasticsearchReindexIT.class.getClassLoader()
+ .getResourceAsStream("activities");
+ List<String> files = IOUtils.readLines(testActivityFolderStream, Charsets.UTF_8);
+
+ for( String file : files) {
+ LOGGER.info("File: " + file );
+ InputStream testActivityFileStream = ElasticsearchReindexIT.class.getClassLoader()
+ .getResourceAsStream("activities/" + file);
+ Activity activity = MAPPER.readValue(testActivityFileStream, Activity.class);
+ StreamsDatum datum = new StreamsDatum(activity, activity.getVerb());
+ datum.getMetadata().put("parent", activity.getObject().getObjectType());
+ datum.getMetadata().put("type", "activity");
+ setupWriter.write( datum );
+ LOGGER.info("Wrote: " + activity.getVerb() );
+ }
+
+ setupWriter.cleanUp();
+
+ flushAndRefresh();
+
+ }
+
+ @Test
+ public void testReindex() throws Exception {
+
+ assert(indexExists("source"));
+ long srcCount = client().count(client().prepareCount("source").request()).get().getCount();
+ assert srcCount > 0;
+
+ ElasticsearchReindexConfiguration reindexConfiguration = MAPPER.readValue(
+ ElasticsearchReindexParentChildIT.class.getResourceAsStream("/testReindexParent.json"), ElasticsearchReindexConfiguration.class);
+
+ reindexConfiguration.getDestination().setClusterName(cluster().getClusterName());
+ reindexConfiguration.getSource().setClusterName(cluster().getClusterName());
+
+ ElasticsearchReindex reindex = new ElasticsearchReindex(reindexConfiguration);
+
+ Thread reindexThread = new Thread(reindex);
+ reindexThread.start();
+ reindexThread.join();
+
+ flushAndRefresh();
+
+ ElasticsearchReindexConfiguration reindexConfiguration2 = MAPPER.readValue(
+ ElasticsearchReindexParentChildIT.class.getResourceAsStream("/testReindexChild.json"), ElasticsearchReindexConfiguration.class);
+
+ reindexConfiguration2.getDestination().setClusterName(cluster().getClusterName());
+ reindexConfiguration2.getSource().setClusterName(cluster().getClusterName());
+
+ ElasticsearchReindex reindex2 = new ElasticsearchReindex(reindexConfiguration2);
+
+ Thread reindexThread2 = new Thread(reindex2);
+ reindexThread2.start();
+ reindexThread2.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/76c26e76/local/elasticsearch-reindex/src/test/resources/ActivityChildObjectParent.json
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/test/resources/ActivityChildObjectParent.json b/local/elasticsearch-reindex/src/test/resources/ActivityChildObjectParent.json
new file mode 100644
index 0000000..963fbe7
--- /dev/null
+++ b/local/elasticsearch-reindex/src/test/resources/ActivityChildObjectParent.json
@@ -0,0 +1,15 @@
+{
+ "template": "*",
+ "order": 100,
+ "mappings": {
+ "object": {
+ "dynamic": true
+ },
+ "activity": {
+ "_parent": {
+ "type": "object"
+ },
+ "dynamic": true
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/76c26e76/local/elasticsearch-reindex/src/test/resources/testReindex.json
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/test/resources/testReindex.json b/local/elasticsearch-reindex/src/test/resources/testReindex.json
new file mode 100644
index 0000000..0cc5aa0
--- /dev/null
+++ b/local/elasticsearch-reindex/src/test/resources/testReindex.json
@@ -0,0 +1,26 @@
+{
+ "source": {
+ "hosts": [
+ "localhost"
+ ],
+ "port": 9300,
+ "clusterName": "elasticsearch",
+ "indexes": [
+ "source"
+ ],
+ "types": [
+ "activity"
+ ]
+ },
+ "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/76c26e76/local/elasticsearch-reindex/src/test/resources/testReindexChild.json
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/test/resources/testReindexChild.json b/local/elasticsearch-reindex/src/test/resources/testReindexChild.json
new file mode 100644
index 0000000..0cc5aa0
--- /dev/null
+++ b/local/elasticsearch-reindex/src/test/resources/testReindexChild.json
@@ -0,0 +1,26 @@
+{
+ "source": {
+ "hosts": [
+ "localhost"
+ ],
+ "port": 9300,
+ "clusterName": "elasticsearch",
+ "indexes": [
+ "source"
+ ],
+ "types": [
+ "activity"
+ ]
+ },
+ "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/76c26e76/local/elasticsearch-reindex/src/test/resources/testReindexParent.json
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/test/resources/testReindexParent.json b/local/elasticsearch-reindex/src/test/resources/testReindexParent.json
new file mode 100644
index 0000000..c601cea
--- /dev/null
+++ b/local/elasticsearch-reindex/src/test/resources/testReindexParent.json
@@ -0,0 +1,26 @@
+{
+ "source": {
+ "hosts": [
+ "localhost"
+ ],
+ "port": 9300,
+ "clusterName": "elasticsearch",
+ "indexes": [
+ "source"
+ ],
+ "types": [
+ "object"
+ ]
+ },
+ "destination": {
+ "hosts": [
+ "localhost"
+ ],
+ "port": 9300,
+ "clusterName": "elasticsearch",
+ "index": "destination",
+ "type": "object",
+ "forceUseConfig": true
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/76c26e76/local/pom.xml
----------------------------------------------------------------------
diff --git a/local/pom.xml b/local/pom.xml
index b75e138..184d463 100644
--- a/local/pom.xml
+++ b/local/pom.xml
@@ -37,6 +37,7 @@
</properties>
<modules>
+ <module>elasticsearch-reindex</module>
</modules>
</project>
[2/2] incubator-streams-examples git commit: incorporated feedback
from https://github.com/apache/incubator-streams-examples/pull/1
Posted by sb...@apache.org.
incorporated feedback from https://github.com/apache/incubator-streams-examples/pull/1
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/a05d4c9b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/tree/a05d4c9b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/diff/a05d4c9b
Branch: refs/heads/master
Commit: a05d4c9bfc300b31fc50ae17dcf513e161a7b13d
Parents: 76c26e7
Author: sblackmon <sb...@apache.org>
Authored: Thu Feb 19 11:13:16 2015 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Thu Feb 19 11:13:16 2015 -0600
----------------------------------------------------------------------
local/elasticsearch-reindex/pom.xml | 19 ++++++++++++++++
.../example/ElasticsearchReindex.java | 23 ++++++++------------
2 files changed, 28 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/a05d4c9b/local/elasticsearch-reindex/pom.xml
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/pom.xml b/local/elasticsearch-reindex/pom.xml
index 4bb6be3..a75fa2a 100644
--- a/local/elasticsearch-reindex/pom.xml
+++ b/local/elasticsearch-reindex/pom.xml
@@ -99,6 +99,21 @@
<version>0.2-incubating-SNAPSHOT</version>
</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>
@@ -152,6 +167,10 @@
<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>
http://git-wip-us.apache.org/repos/asf/incubator-streams-examples/blob/a05d4c9b/local/elasticsearch-reindex/src/main/java/org/apache/streams/elasticsearch/example/ElasticsearchReindex.java
----------------------------------------------------------------------
diff --git a/local/elasticsearch-reindex/src/main/java/org/apache/streams/elasticsearch/example/ElasticsearchReindex.java b/local/elasticsearch-reindex/src/main/java/org/apache/streams/elasticsearch/example/ElasticsearchReindex.java
index 37e178c..dc94773 100644
--- a/local/elasticsearch-reindex/src/main/java/org/apache/streams/elasticsearch/example/ElasticsearchReindex.java
+++ b/local/elasticsearch-reindex/src/main/java/org/apache/streams/elasticsearch/example/ElasticsearchReindex.java
@@ -44,41 +44,36 @@ public class ElasticsearchReindex implements Runnable {
private final static Logger LOGGER = LoggerFactory.getLogger(ElasticsearchReindex.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());
- }
-
- ElasticsearchReindexConfiguration reindex;
+ ElasticsearchReindexConfiguration config;
public ElasticsearchReindex() {
- this(new ComponentConfigurator<ElasticsearchReindexConfiguration>(ElasticsearchReindexConfiguration.class).detectConfiguration(StreamsConfigurator.getConfig()));
+ this(new ComponentConfigurator<>(ElasticsearchReindexConfiguration.class).detectConfiguration(StreamsConfigurator.getConfig()));
}
public ElasticsearchReindex(ElasticsearchReindexConfiguration reindex) {
- this.reindex = reindex;
+ this.config = reindex;
}
public static void main(String[] args)
{
LOGGER.info(StreamsConfigurator.config.toString());
- executor.submit(new ElasticsearchReindex());
+ ElasticsearchReindex reindex = new ElasticsearchReindex();
+
+ new Thread(reindex).start();
}
@Override
public void run() {
- ElasticsearchPersistReader elasticsearchPersistReader = new ElasticsearchPersistReader(reindex.getSource());
+ ElasticsearchPersistReader elasticsearchPersistReader = new ElasticsearchPersistReader(config.getSource());
- ElasticsearchPersistWriter elasticsearchPersistWriter = new ElasticsearchPersistWriter(reindex.getDestination());
+ 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);