java - 在Apache Camel中停止SQL路由

Question

我目前正在尝试使用Apache Camel将一些数据从Oracle导入到ElasticSearch(以JSON格式)。我是这个框架的新手，所以我想您可能会帮忙的!

这是我的代码:

from("sql://select log_instance_id, calculated_status, begin_date, end_date, mediation_descr_code, duration from of_log_mediation_instance?dataSource=myDataSource")
.to("elasticsearch://boa-elasticsearch?operation=INDEX&indexName=twitter&indexType=tweet")

问题是该路由永远不会停止，因为我使用from(“sql:select ...”)对其进行了调用，它第一次完成，然后重新运行该路由，直到我在Thread.sleep(10000)的末尾停止上下文为止。 )。我尝试从其他端点开始，但是我经常遇到此错误:

org.apache.camel.TypeConversionException: Error during type conversion from type: java.lang.String to the required type: org.elasticsearch.action.index.IndexRequest with value [{LOG_INSTANCE_ID=4029927, CALCULATED_STATUS=flowError, BEGIN_DATE=2015-05-21 16:54:31.45436, END_DATE=2015-05-21 16:54:31.495735, MEDIATION_DESCR_CODE=DiffuserMAJOfrCial, DURATION=0.041}, {LOG_INSTANCE_ID=4140006, CALCULATED_STATUS=flowCompleted, BEGIN_DATE=2015-06-01 11:11:31.569414, END_DATE=2015-06-01 11:11:33.109604, MEDIATION_DESCR_CODE=RecevoirDemOpeContrat, DURATION=1.54}, {LOG_INSTANCE_ID=4038409, CALCULATED_STATUS=flowError, BEGIN_DATE=2015-05-22 10:18:11.295776, END_DATE=2015-05-22 10:18:11.299125, MEDIATION_DESCR_CODE=DiffuserMAJOfrCial, DURATION=0.004}, {LOG_INSTANCE_ID=4144321, CALCULATED_STATUS=flowCompleted, BEGIN_DATE=2015-06-01 15:33:12.37841, END_DATE=2015-06-01 15:33:12.822529, MEDIATION_DESCR_CODE=RecevoirDemOpeContrat, DURATION=0.444}, {LOG_INSTANCE_ID=4039159, CALCULATED_STATUS=flowError, BEGIN_DATE=2015-05-22 11:23:56.462823, END_DATE=2015-05-22 11:23:58.15667, MEDIATION_DESCR_CODE=DiffuserMAJOfrCial, DURATION=1.694}, {LOG_INSTANCE_ID=4045264, CALCULATED_STATUS=... [Body clipped after 1000 chars, total length is 16546230] due java.lang.NullPointerException
    at org.apache.camel.impl.converter.BaseTypeConverterRegistry.createTypeConversionException(BaseTypeConverterRegistry.java:571)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.impl.converter.BaseTypeConverterRegistry.convertTo(BaseTypeConverterRegistry.java:129)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.impl.MessageSupport.getBody(MessageSupport.java:72)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.impl.MessageSupport.getBody(MessageSupport.java:47)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.component.elasticsearch.ElasticsearchProducer.process(ElasticsearchProducer.java:116)[camel-elasticsearch-2.15.2.jar:2.15.2]
    at org.apache.camel.util.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:61)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:129)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:118)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:80)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.loadbalancer.QueueLoadBalancer.process(QueueLoadBalancer.java:44)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:109)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.processor.loadbalancer.LoadBalancerSupport.process(LoadBalancerSupport.java:87)[camel-core-2.15.2.jar:2.15.2]
    at org.apache.camel.component.quartz2.CamelJob.execute(CamelJob.java:56)[camel-quartz2-2.15.2.jar:2.15.2]
    at org.quartz.core.JobRunShell.run(JobRunShell.java:202)[quartz-2.2.1.jar:]
    at org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:573)[quartz-2.2.1.jar:]
[artzScheduler-camel-1_Worker-1] CamelJob                       ERROR Error processing exchange. Exchange[Message: [{LOG_INSTANCE_ID=4029927, CALCULATED_STATUS=flowError, BEGIN_DATE=2015-05-21 16:54:31.45436, END_DATE=2015-05-21 16:54:31.495735, MEDIATION_DESCR_CODE=DiffuserMAJOfrCial, DURATION=0.041}, {LOG_INSTANCE_ID=4140006, CALCULATED_STATUS=flowCompleted, BEGIN_DATE=2015-06-01 11:11:31.569414, END_DATE=2015-06-01 11:11:33.109604, MEDIATION_DESCR_CODE=RecevoirDemOpeContrat, DURATION=1.54}, {LOG_INSTANCE_ID=4038409, CALCULATED_STATUS=flowError, BEGIN_DATE=2015-05-22 10:18:11.295776, END_DATE=2015-05-22 10:18:11.299125, MEDIATION_DESCR_CODE=DiffuserMAJOfrCial, DURATION=0.004}, {LOG_INSTANCE_ID=4144321, CALCULATED_STATUS=flowCompleted, BEGIN_DATE=2015-06-01 15:33:12.37841, END_DATE=2015-06-01 15:33:12.822529, MEDIATION_DESCR_CODE=RecevoirDemOpeContrat, DURATION=0.444}, {LOG_INSTANCE_ID=4039159, CALCULATED_STATUS=flowError, BEGIN_DATE=2015-05-22 11:23:56.462823, END_DATE=2015-05-22 11:23:58.15667, MEDIATION_DESCR_CODE=DiffuserMAJOfrCial, DURATION=1.694}, {LOG_INSTANCE_ID=4045264, CALCULATED_STATUS=... [Body clipped after 1000 chars, total length is 16546230]]. Caused by: [org.quartz.JobExecutionException - org.apache.camel.TypeConversionException: Error during type conversion from type: java.lang.String to the required type: org.elasticsearch.action.index.IndexRequest with value [{LOG_INSTANCE_ID=4029927, CALCULATED_STATUS=flowError, BEGIN_DATE=2015-05-21 16:54:31.45436, END_DATE=2015-05-21 16:54:31.495735, MEDIATION_DESCR_CODE=DiffuserMAJOfrCial, DURATION=0.041}, {LOG_INSTANCE_ID=4140006, CALCULATED_STATUS=flowCompleted, BEGIN_DATE=2015-06-01 11:11:31.569414, END_DATE=2015-06-01 11:11:33.109604, MEDIATION_DESCR_CODE=RecevoirDemOpeContrat, DURATION=1.54}, {LOG_INSTANCE_ID=4038409, CALCULATED_STATUS=flowError, BEGIN_DATE=2015-05-22 10:18:11.295776, END_DATE=2015-05-22 10:18:11.299125, MEDIATION_DESCR_CODE=DiffuserMAJOfrCial, DURATION=0.004}, {LOG_INSTANCE_ID=4144321, CALCULATED_STATUS=flowCompleted, BEGIN_DATE=2015-06-01 15:33:12.37841, END_DATE=2015-06-01 15:33:12.822529, MEDIATION_DESCR_CODE=RecevoirDemOpeContrat, DURATION=0.444}, {LOG_INSTANCE_ID=4039159, CALCULATED_STATUS=flowError, BEGIN_DATE=2015-05-22 11:23:56.462823, END_DATE=2015-05-22 11:23:58.15667, MEDIATION_DESCR_CODE=DiffuserMAJOfrCial, DURATION=1.694}, {LOG_INSTANCE_ID=4045264, CALCULATED_STATUS=... [Body clipped after 1000 chars, total length is 16546230] due java.lang.NullPointerException]

我试过直接 quartz quartz scheluder，.end()，loop(1)，...

谢谢 !

Answer 1

如果您需要每X次运行一次，请尝试使用计时器组件

from("timer://foo?fixedRate=true&period=60000")
    .to("sql://select log_instance_id, calculated_status, begin_date, end_date, mediation_descr_code, duration from of_log_mediation_instance?dataSource=myDataSource")
    .to("elasticsearch://boa-elasticsearch?operation=INDEX&indexName=twitter&indexType=tweet")

该路线将每1分钟运行一次。如果您需要更复杂的逻辑，请尝试使用 quartz 组件

from("quartz://foo/bar?cron=0+0/1+*+*+*+?")
    .to("sql://select log_instance_id, calculated_status, begin_date, end_date, mediation_descr_code, duration from of_log_mediation_instance?dataSource=myDataSource")
    .to("elasticsearch://boa-elasticsearch?operation=INDEX&indexName=twitter&indexType=tweet")