Как избежать тайм-аута обмена в адаптере RedHat Fuse, работающем с Apache Camel и CXF? ⇐ JAVA

[Anonymous]

Я работаю с адаптером RedHat Fuse в Java DSL, используя Apache Camel и CXF. Я пытаюсь обрабатывать относительно большие входящие сообщения SOAP, но у меня возникла проблема с тайм-аутом. Затем отправитель получает ошибку 500 Internal Server Error, хотя обработка с нашей стороны проходит нормально. Иногда время обработки немного быстрее, тогда отправляющая сторона получает 200, и все в порядке.
Похоже, что у CXF или Camel возникает какой-то тайм-аут («Сообщение OUT ' занимает слишком много времени), но для нас это происходит слишком быстро. Кажется, проблема в том, что я не могу найти, где и как установлен этот тайм-аут в 30000 мс и как его изменить.
Любая помощь или указания по CXF или Camel будут оценены здесь, например, что именно представляет собой этот тайм-аут или какое свойство его меняет.
Ниже показано сообщение об ошибке, которое я получаю, и маршрут, который я использую:

Код: Выделить всё

26-09-24 07:52:12.575 [XNIO-1 task-1] WARN  o.a.cxf.phase.PhaseInterceptorChain - Application {http://example.namespace.com/wsdl/Service/V1}MyService#{http://example.namespace.com/wsdl/Service/V1}processRequest has thrown exception, unwinding now
org.apache.cxf.interceptor.Fault: The OUT message was not received within: 30000 millis. Exchange[91F3A9AAC0D6092-0000000000000000]
at org.apache.camel.component.cxf.jaxws.CxfConsumer$CxfConsumerInvoker.checkFailure(CxfConsumer.java:350)
at org.apache.camel.component.cxf.jaxws.CxfConsumer$CxfConsumerInvoker.setResponseBack(CxfConsumer.java:322)
at org.apache.camel.component.cxf.jaxws.CxfConsumer$CxfConsumerInvoker.asyncInvoke(CxfConsumer.java:212)
at org.apache.camel.component.cxf.jaxws.CxfConsumer$CxfConsumerInvoker.invoke(CxfConsumer.java:161)
...
Caused by: org.apache.camel.ExchangeTimedOutException: The OUT message was not received within: 30000 millis. Exchange[91F3A9AAC0D6092-0000000000000000]
at org.apache.camel.component.cxf.jaxws.CxfConsumer$CxfConsumerInvoker.asyncInvoke(CxfConsumer.java:210)
... 105 common frames omitted

А вот маршрут, который я использую (Java DSL):

Код: Выделить всё

package com.example.routes;

import com.example.logging.CustomLogger;
import com.example.beans.TransformData;
import com.example.beans.UUIDGenerator;
import com.example.configuration.ExceptionConnectionErrorResolver;
import com.example.webservices.WebServiceConfig;
import org.apache.camel.Exchange;
import org.apache.camel.LoggingLevel;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.builder.ThreadPoolBuilder;
import org.apache.camel.CamelContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.concurrent.ExecutorService;

@Component
public class MyServiceRoute extends RouteBuilder {
@Autowired
CustomLogger customLogger;

@Autowired
WebServiceConfig webServiceConfig;

@Autowired
UUIDGenerator generator;

@Autowired
ExceptionConnectionErrorResolver exceptionConnectionErrorResolver;

@Override
public void configure() throws Exception {

CamelContext context = getContext();

ExecutorService executorService = new ThreadPoolBuilder(context)
.poolSize(150)
.maxPoolSize(150)
.maxQueueSize(-1)
.build("SplitThreadPool");

from("cxf:bean:processRequestSoapEndpoint?exchangePattern=InOnly").routeId("myServiceRoute")

.onException(org.apache.camel.support.processor.validation.SchemaValidationException.class,org.xml.sax.SAXParseException.class,org.apache.camel.ValidationException.class).id("onTransformationException")
.handled(true)
.bean(customLogger, "logCode(${exchangeProperty.BreadCrumb}, CODE-992, Validation of message failed.  ${exception.message})").id("logCode992")
.process(exchange -> {
Exception exception = exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Exception.class);
exchange.getContext().createProducerTemplate()
.sendBody("log:error?showAll=true&multiline=true",
"CODE-992 Validation Error: " + exception.getMessage() + "\nStack Trace:\n"  + getStackTrace(exception));
})
.end()

.onException(javax.xml.transform.TransformerException.class).id("OnValidationException")
.handled(true)
.bean(customLogger, "logCode(${exchangeProperty.BreadCrumb}, CODE-032)").id("logCode032")
.end()

.onException(org.apache.cxf.interceptor.Fault.class,org.apache.camel.http.base.HttpOperationFailedException.class).id("onEndpointException")
.handled(true)
.bean(customLogger, "logCode(${exchangeProperty.BreadCrumb},CODE-043)").id("logCode043")
.bean(customLogger, "logCode(${exchangeProperty.BreadCrumb},CODE-996,Error: ${exception.message})").id("logCode996")
.end()

.onException(java.net.ConnectException.class,org.apache.http.conn.HttpHostConnectException.class).id("onConnectionEndpoint")
.handled(true)
.maximumRedeliveries(exceptionConnectionErrorResolver.getMaxRedeliveries())
.redeliveryDelay(exceptionConnectionErrorResolver.getRedeliveryDelay())
.retryAttemptedLogLevel(LoggingLevel.WARN)
.bean(customLogger, String.format("logCode(${exchangeProperty.BreadCrumb},CODE-042,%s)",webServiceConfig.processRequestSoapEndpointBackend().getAddress())).id("logCode042")
.bean(customLogger, "logCode(${exchangeProperty.BreadCrumb},CODE-996,Error: ${exception.message})").id("logCode996")
.end()

.onException(java.lang.Exception.class).id("onGeneralException")
.handled(true)
.bean(customLogger, "logCode(${exchangeProperty.BreadCrumb},CODE-000)").id("logCode000")
.bean(customLogger, "logCode(${exchangeProperty.BreadCrumb},CODE-996,Error:  ${exception.message})").id("logUnexpectedException")
.end()

.bean(customLogger, "logStart").id("logStart")
.bean("PropertyGenerator").id("setProperties")
.setProperty("messageHeaderProperty", xpath("/ns:processRequestHeader", XpathNameSpaces.namespaces)).id("setMessageHeaderProperty")
.setProperty("voorloopRecordProperty", xpath("/ns:processRequestBody/ns:businessData/ns:record", XpathNameSpaces.namespaces)).id("setVoorloopRecordProperty")
.to("validator:classpath:services/ExampleService/xsd/BM/V1/ExampleMessage.xsd").id("validateExampleMessage")
.bean(customLogger, "logCode(${exchangeProperty.BreadCrumb}, CODE-092)").id("logCode092")
.split(xpath("/ns:processRequestBody/ns:businessData/ns:record/ns:person", XpathNameSpaces.namespaces)).parallelProcessing().executorService(executorService).stopOnException().id("splitPerPerson")
.bean(TransformData.class).id("setSplittedBody")
.process(generator).id("setUUIDProp")
.setProperty("timestampProperty").simple("${date:now:yyyy-MM-dd'T'HH:mm:ss.SSS'Z'}").id("setTimestamp")
.to("xslt:classpath:/transformations/ExampleTransform.xsl").id("transformExample")
.to("validator:classpath:services/ExampleService/xsd/BAS/V1/ExampleValidation.xsd").id("validateExampleValidation")
.removeHeaders("*", "BreadCrumb").id("removeHeaders")
.to("cxf:bean:processRequestSoapEndpointBackend").id("toSoapEndpoint")
.bean(customLogger, String.format("logCode(${exchangeProperty.BreadCrumb}, CODE-041,%s)",webServiceConfig.processRequestSoapEndpointBackend().getAddress())).id("logCode041")
.bean(customLogger, "logEnd").id("logCodeEnd");
}

private String getStackTrace(Exception exception) {
StringWriter sw = new StringWriter();
PrintWriter pw = new PrintWriter(sw);
exception.printStackTrace(pw);
return sw.toString();
}
}

Мы пытались ускорить время обработки с помощью параллельной обработки и небольших изменений, чтобы попытаться сократить время до 30 секунд, но реальная проблема заключается в том, что этот тайм-аут слишком мал.
Проблема видимо в том, что я не могу найти где и как выставлен этот таймаут 30000мс, и как его изменить. Я пробовал некоторые вещи, например cxf.timeout=60000, но, похоже, у CXF нет этого свойства.
Такие вещи, как Cxf.ConnectionTimeOut и Cxf. Кажется, что ReceiveTimeout относится к тайм-ауту HTTP, но я думаю, что это какой-то внутренний тайм-аут приложения.

Подробнее здесь: https://stackoverflow.com/questions/790 ... he-camel-a

[Alex]

Привет, скорее всего это параметр continuationTimeout, по умолчанию он равен 30000 мс.
Эта опция используется для установки тайм-аута продолжения работы CXF, который может использоваться в CxfConsumer по умолчанию, если сервер CXF использует транспорт Jetty или Servlet.

from("cxf:bean:processRequestSoapEndpoint?exchangePattern=InOnly&continuationTimeout=60000")