maven - 我如何在Apache Camel 中使用Elasticsearch 2.2.1

Question

我正在尝试开发一个Apache Camel应用程序，该应用程序使用来自Amazon SQS Queue的消息，然后将其发送到elasticsearch。
这是我的pom.xml文件:

   <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">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.nsm.sam</groupId>
     <artifactId>CEP_CAMEL</artifactId>
     <version>0.0.1</version>
     <dependencies>
           <dependency>
               <groupId>org.apache.camel</groupId>
               <artifactId>camel-core</artifactId>
               <version>2.16.3</version>
           </dependency>
           <dependency>
               <groupId>org.slf4j</groupId>
               <artifactId>slf4j-simple</artifactId>
               <version>1.7.5</version>
           </dependency>
           <dependency>
              <groupId>org.apache.camel</groupId>
              <artifactId>camel-aws</artifactId>
<!--            <version>2.8.0</version> -->
              <version>2.16.3</version>
          </dependency>
            <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-google-mail</artifactId>
            <version>2.16.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-mail</artifactId>
            <version>2.16.3</version>
        </dependency>
        <dependency>
            <groupId>org.apache.camel</groupId>
            <artifactId>camel-gae</artifactId>
            <version>2.16.3</version>
        </dependency>
        <dependency>
          <groupId>org.apache.camel</groupId>
          <artifactId>camel-jsonpath</artifactId>
          <version>2.16.3</version>
        </dependency>
        <dependency>
        <groupId>org.apache.camel</groupId>
            <artifactId>camel-elasticsearch</artifactId>
            <version>2.16.3</version>
        </dependency>

        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>2.2.1</version>
        </dependency>

       </dependencies>
   </project>

和我的主要应用程序:

package com.nsm.sam.cep;

  import java.util.Scanner;

  import org.apache.camel.CamelContext;
  import org.apache.camel.builder.RouteBuilder;
  import org.apache.camel.impl.DefaultCamelContext;
  import org.apache.camel.impl.SimpleRegistry;

  import com.amazonaws.ClientConfiguration;
  import com.amazonaws.auth.AWSCredentials;
  import com.amazonaws.auth.BasicAWSCredentials;
  import com.amazonaws.services.sqs.AmazonSQSClient;
  import com.amazonaws.util.json.JSONException;

  public class CEPBroker {
    public static void main(String args[]) throws Exception {
        final SimpleRegistry registry = new SimpleRegistry(); 
        AWSCredentials awsCredentials = new   BasicAWSCredentials("nnnnnnnnnnnnnnmy acces key", "xxxxxxxxxxxxxx my   secret key");
        ClientConfiguration clientConfiguration = new   ClientConfiguration(); 

        final AmazonSQSClient client = new AmazonSQSClient(awsCredentials,   clientConfiguration);
        client.setEndpoint("url .....");
        registry.put("amazonclient" , client); 
        final CamelContext context = new DefaultCamelContext(registry);
        context.addRoutes(new RouteBuilder() {
            public void configure() throws JSONException {
                from("aws-sqs://SAM_SQS_test?amazonSQSClient=#amazonclient")
                //.resequence().body()
                //.process(new MesssageProcessor());
                .process(new TypeProcessor())
                .to("elasticsearch://elasticsearch?operation=INDEX&indexName=sam_monitoring&indexType=login&port=9300&ip=127.0.0.1");

            }
        });
        context.start();
        System.out.println("Press enter to stop CAMEL ...");
        Scanner keyboard = new Scanner(System.in);
        keyboard.nextLine();
        context.stop();
    }
  }

我安装的Elasticsearch版本是2.2.1

当我用elasticsearch 1.7.5尝试时，它正在工作
但是有2.2.1

我有这个异常(exception)

[main] INFO org.apache.camel.impl.DefaultCamelContext - Apache Camel     2.16.3 (CamelContext: camel-1) is starting
  [main] INFO org.apache.camel.management.ManagedManagementStrategy - JMX is enabled
  [main] INFO org.apache.camel.impl.converter.DefaultTypeConverter - Loaded 198 type converters
  [main] INFO org.apache.camel.impl.DefaultRuntimeEndpointRegistry -   Runtime endpoint registry is in extended mode gathering usage statistics   of all incoming and outgoing endpoints (cache limit: 1000)
  Exception in thread "main" java.lang.NoClassDefFoundError:   org/elasticsearch/action/support/replication/ReplicationType
    at org.apache.camel.component.elasticsearch.ElasticsearchConstants.  <clinit>(ElasticsearchConstants.java:46)
    at   org.apache.camel.component.elasticsearch.ElasticsearchConfiguration.<init>  (ElasticsearchConfiguration.java:43)
    at   org.apache.camel.component.elasticsearch.ElasticsearchComponent.createEndp  oint(ElasticsearchComponent.java:43)
    at   org.apache.camel.impl.DefaultComponent.createEndpoint(DefaultComponent.jav  a:114)
      at   org.apache.camel.impl.DefaultCamelContext.getEndpoint(DefaultCamelContext.  java:567)
    at   org.apache.camel.util.CamelContextHelper.getMandatoryEndpoint(CamelContext  Helper.java:79)
    at   org.apache.camel.model.RouteDefinition.resolveEndpoint(RouteDefinition.jav  a:211)
    at   org.apache.camel.impl.DefaultRouteContext.resolveEndpoint(DefaultRouteCont  ext.java:107)
    at org.apache.camel.impl.DefaultRouteContext.resolveEndpoint(DefaultRouteCont  ext.java:113)
    at org.apache.camel.model.SendDefinition.resolveEndpoint(SendDefinition.java:62)
    at org.apache.camel.model.SendDefinition.createProcessor(SendDefinition.java:56)
    at org.apache.camel.model.ProcessorDefinition.makeProcessorImpl(ProcessorDefinition.java:535)
    at org.apache.camel.model.ProcessorDefinition.makeProcessor(ProcessorDefinition.java:496)
    at org.apache.camel.model.ProcessorDefinition.addRoutes(ProcessorDefinition.java:220)
    at org.apache.camel.model.RouteDefinition.addRoutes(RouteDefinition.java:1069)
    at org.apache.camel.model.RouteDefinition.addRoutes(RouteDefinition.java:196)
    at org.apache.camel.impl.DefaultCamelContext.startRoute(DefaultCamelContext.java:944)
    at org.apache.camel.impl.DefaultCamelContext.startRouteDefinitions(DefaultCamelContext.java:3260)
    at org.apache.camel.impl.DefaultCamelContext.doStartCamel(DefaultCamelContext.java:2983)
    at org.apache.camel.impl.DefaultCamelContext.access$000(DefaultCamelContext.java:175)
    at org.apache.camel.impl.DefaultCamelContext$2.call(DefaultCamelContext.java:2814)
    at org.apache.camel.impl.DefaultCamelContext$2.call(DefaultCamelContext.java:2810)
    at   org.apache.camel.impl.DefaultCamelContext.doWithDefinedClassLoader(DefaultCamelContext.java:2833)
    at   org.apache.camel.impl.DefaultCamelContext.doStart(DefaultCamelContext.java  :2810)
    at  rg.apache.camel.support.ServiceSupport.start(ServiceSupport.java:61)
    at org.apache.camel.impl.DefaultCamelContext.start(DefaultCamelContext.java:2779)
    at com.nsm.sam.cep.CEPBroker.main(CEPBroker.java:43)
  Caused by: java.lang.ClassNotFoundException:   org.elasticsearch.action.support.replication.ReplicationType
      at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    ... 27 more

Answer 1

我自己碰到了这个确切的问题。不幸的是， Camel 2.16.x不支持> = 2.0的elasticsearch

但是，我可以确认 Camel 2.17.0确实可以与Elasticsearch 2.3.1一起使用。

Camel release notes表示 Camel 2.17.0现在支持elasticsearch 2.0。

Elasticsearch from 1.7.3 to 2.0.0