Как исправить проблему с потоком в задании потока данных из стандартной среды GAE? ⇐ JAVA

[Anonymous]

Я хочу использовать поток данных из приложения Google App Engine (в стандартной среде) в Java 21.
Но я получаю следующую ошибку:

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

java.lang.RuntimeException: Error while staging packages
at org.apache.beam.runners.dataflow.util.PackageUtil.stageClasspathElements(PackageUtil.java:372)
at org.apache.beam.runners.dataflow.util.PackageUtil.stageClasspathElements(PackageUtil.java:238)
at org.apache.beam.runners.dataflow.util.GcsStager.stageFiles(GcsStager.java:53)
at org.apache.beam.runners.dataflow.DataflowRunner.stageArtifacts(DataflowRunner.java:1068)
at org.apache.beam.runners.dataflow.DataflowRunner.run(DataflowRunner.java:1217)
at org.apache.beam.runners.dataflow.DataflowRunner.run(DataflowRunner.java:203)
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:325)
at org.apache.beam.sdk.Pipeline.run(Pipeline.java:310)
...
at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)
at java.base/java.lang.reflect.Method.invoke(Method.java:580)
at org.glassfish.jersey.server.model.internal.ResourceMethodInvocationHandlerFactory.lambda$static$0(ResourceMethodInvocationHandlerFactory.java:52)
at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher$1.run(AbstractJavaResourceMethodDispatcher.java:146)
at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.invoke(AbstractJavaResourceMethodDispatcher.java:189)
at org.glassfish.jersey.server.model.internal.JavaResourceMethodDispatcherProvider$ResponseOutInvoker.doDispatch(JavaResourceMethodDispatcherProvider.java:176)
at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.dispatch(AbstractJavaResourceMethodDispatcher.java:93)
at org.glassfish.jersey.server.model.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:478)
at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:400)
at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:81)
at org.glassfish.jersey.server.ServerRuntime$1.run(ServerRuntime.java:274)
at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248)
at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244)
at org.glassfish.jersey.internal.Errors.process(Errors.java:292)
at org.glassfish.jersey.internal.Errors.process(Errors.java:274)
at org.glassfish.jersey.internal.Errors.process(Errors.java:244)
at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:266)
at org.glassfish.jersey.server.ServerRuntime.process(ServerRuntime.java:253)
at org.glassfish.jersey.server.ApplicationHandler.handle(ApplicationHandler.java:696)
at org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:397)
at org.glassfish.jersey.servlet.WebComponent.service(WebComponent.java:349)
at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:358)
at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:312)
at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:205)
at org.eclipse.jetty.ee10.servlet.ServletHolder$NotAsync.service(ServletHolder.java:1379)
at org.eclipse.jetty.ee10.servlet.ServletHolder.handle(ServletHolder.java:736)
at org.eclipse.jetty.ee10.servlet.ServletHandler$ChainEnd.doFilter(ServletHandler.java:1614)
at com.google.apphosting.utils.servlet.ee10.JdbcMySqlConnectionCleanupFilter.doFilter(JdbcMySqlConnectionCleanupFilter.java:78)
at org.eclipse.jetty.ee10.servlet.FilterHolder.doFilter(FilterHolder.java:205)
at org.eclipse.jetty.ee10.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1586)
at  ... MyFilter
at org.eclipse.jetty.ee10.servlet.FilterHolder.doFilter(FilterHolder.java:205)
at org.eclipse.jetty.ee10.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1586)
at  ... MyFilter2
at org.eclipse.jetty.ee10.servlet.FilterHolder.doFilter(FilterHolder.java:205)
at org.eclipse.jetty.ee10.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1586)
at  ...  MyFilter3
at org.eclipse.jetty.ee10.servlet.FilterHolder.doFilter(FilterHolder.java:205)
at org.eclipse.jetty.ee10.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1586)
at com.google.apphosting.runtime.jetty.ee10.ParseBlobUploadFilter.doFilter(ParseBlobUploadFilter.java:120)
at org.eclipse.jetty.ee10.servlet.FilterHolder.doFilter(FilterHolder.java:208)
at org.eclipse.jetty.ee10.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1586)
at org.eclipse.jetty.ee10.servlet.ServletHandler$MappedServlet.handle(ServletHandler.java:1547)
at org.eclipse.jetty.ee10.servlet.ServletChannel.dispatch(ServletChannel.java:824)
at org.eclipse.jetty.ee10.servlet.ServletChannel.handle(ServletChannel.java:436)
at org.eclipse.jetty.ee10.servlet.ServletHandler.handle(ServletHandler.java:464)
at org.eclipse.jetty.security.SecurityHandler.handle(SecurityHandler.java:575)
at org.eclipse.jetty.ee10.servlet.SessionHandler.handle(SessionHandler.java:703)
at org.eclipse.jetty.server.handler.ContextHandler.handle(ContextHandler.java:1060)
at com.google.apphosting.runtime.jetty.ee10.AppEngineWebAppContext.handle(AppEngineWebAppContext.java:301)
at org.eclipse.jetty.server.handler.HotSwapHandler.handle(HotSwapHandler.java:90)
at com.google.apphosting.runtime.jetty.AppVersionHandler.handle(AppVersionHandler.java:84)
at org.eclipse.jetty.server.Handler$Wrapper.handle(Handler.java:740)
at com.google.apphosting.runtime.jetty.CoreSizeLimitHandler.handle(CoreSizeLimitHandler.java:75)
at org.eclipse.jetty.server.Server.handle(Server.java:181)
at org.eclipse.jetty.server.internal.HttpChannelState$HandlerInvoker.run(HttpChannelState.java:661)
at com.google.apphosting.runtime.jetty.delegate.internal.DelegateConnection.lambda$handle$0(DelegateConnection.java:146)
at com.google.apphosting.runtime.jetty.JettyServletEngineAdapter$2.run(JettyServletEngineAdapter.java:128)
at com.google.apphosting.runtime.jetty.delegate.internal.DelegateConnection.handle(DelegateConnection.java:142)
at com.google.apphosting.runtime.jetty.delegate.DelegateConnector.service(DelegateConnector.java:49)
at com.google.apphosting.runtime.jetty.JettyServletEngineAdapter.serviceRequest(JettyServletEngineAdapter.java:255)
at com.google.apphosting.runtime.RequestRunner.dispatchServletRequest(RequestRunner.java:349)
at com.google.apphosting.runtime.RequestRunner.dispatchRequest(RequestRunner.java:234)
at com.google.apphosting.runtime.RequestRunner.run(RequestRunner.java:200)
at com.google.apphosting.runtime.ThreadGroupPool$PoolEntry.run(ThreadGroupPool.java:273)
at java.base/java.lang.Thread.run(Thread.java:1583)
caused by: java.lang.NullPointerException: Operation not allowed in a thread that is neither the original request thread nor a thread created by ThreadManager
at com.google.apphosting.runtime.ApiProxyImpl$CurrentRequestThreadFactory.newThread(ApiProxyImpl.java:1368)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.(ThreadPoolExecutor.java:637)
at java.base/java.util.concurrent.ThreadPoolExecutor.addWorker(ThreadPoolExecutor.java:928)
at java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1364)
at org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.MoreExecutors$ListeningDecorator.execute(MoreExecutors.java:640)
at java.base/java.util.concurrent.CompletableFuture.asyncRunStage(CompletableFuture.java:1818)
at java.base/java.util.concurrent.CompletableFuture.runAsync(CompletableFuture.java:2033)
at org.apache.beam.sdk.util.MoreFutures.supplyAsync(MoreFutures.java:105)
at org.apache.beam.runners.dataflow.util.PackageUtil.stagePackage(PackageUtil.java:141)
at org.apache.beam.runners.dataflow.util.PackageUtil.lambda$stageClasspathElements$2(PackageUtil.java:324)
at java.base/java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1150)
at java.base/java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482)
...  1 common frames elided

Может ли кто-нибудь мне помочь? Как я могу запустить задание потока данных в соответствии с политикой GAE без потоков?
Вот выдержка из моего pom:

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

org.apache.beam
beam-sdks-java-core
2.59.0


org.apache.beam
beam-runners-google-cloud-dataflow-java
2.59.0


org.apache.beam
beam-sdks-java-io-google-cloud-platform
2.59.0

Вот мой код для запуска задания. Он запускается в асинхронной задаче (в очереди). Цель состоит в том, чтобы обновить все объекты одного типа в хранилище данных:

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

DataflowPipelineOptions dataflowOptions = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
dataflowOptions.setRunner(DataflowRunner.class);
dataflowOptions.setProject("myproject");
dataflowOptions.setRegion("europe-west1");
String dataflowBucket = "mybucket";
dataflowOptions.setTempLocation("gs://" + dataflowBucket + "/tmp");
dataflowOptions.setStagingLocation("gs://" + dataflowBucket + "/staging");

Pipeline pipeline = Pipeline.create(dataflowOptions);

// Step 2: Read from Google Cloud Datastore
log.info("step 2 : Read from Google Cloud Datastore");
String projectId = AppEngine.ID;
String entityKind = "MyEntity";
PCollection entities = pipeline.apply(
"ReadFromDatastore",
DatastoreIO.v1().read().withProjectId(projectId)
.withLiteralGqlQuery("SELECT * FROM " + entityKind)
);

// Step 3: Modify entities
log.info("step 3 : modify entities");
PCollection modifiedEntities = entities.apply(
"ModifyEntities",
ParDo.of(new DoFn() {
@ProcessElement
public void processElement(ProcessContext c) {
Entity entity = c.element();
// Create a mutable version of the entity
Entity.Builder modifiedEntityBuilder = Entity.newBuilder(entity);

// Modify some properties
String fieldName = "name";
String defaultValue = "defaultName";
Value oldValue = entity.getPropertiesOrDefault(fieldName, makeValue(defaultValue).build());
log.info("step 3 : oldValue = " + oldValue.getStringValue());
String newValue = oldValue.getStringValue() + "_modified";
modifiedEntityBuilder.putProperties(fieldName, makeValue(newValue).build());

// Emit the modified entity
c.output(modifiedEntityBuilder.build());
}
})
);

// Step 4: Write modified entities back to Datastore
log.info("step 4 : Write modified entities back to Datastore");

modifiedEntities.apply(
"WriteToDatastore",
DatastoreIO.v1().write().withProjectId(projectId)
);

// Step 5: Run the pipeline
log.info("step 5 : Run the pipeline - begin");
pipeline.run().waitUntilFinish();
log.info("step 5 : Run the pipeline - end");

Я нашел сообщение, в котором у кого-то возникла такая же проблема, но ответа нет.

Подробнее здесь: https://stackoverflow.com/questions/790 ... nvironment