[
https://issues.apache.org/jira/browse/BEAM-4523?focusedWorklogId=110196&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-110196
]
ASF GitHub Bot logged work on BEAM-4523:
----------------------------------------
Author: ASF GitHub Bot
Created on: 08/Jun/18 18:05
Start Date: 08/Jun/18 18:05
Worklog Time Spent: 10m
Work Description: jkff commented on a change in pull request #5588:
[BEAM-4523] Implement batch flink executable stage context
URL: https://github.com/apache/beam/pull/5588#discussion_r194138024
##########
File path:
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/BatchFlinkExecutableStageContext.java
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.
+ */
+package org.apache.beam.runners.flink.translation.functions;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalNotification;
+import org.apache.beam.runners.core.construction.graph.ExecutableStage;
+import org.apache.beam.runners.flink.ArtifactSourcePool;
+import org.apache.beam.runners.fnexecution.control.DockerJobBundleFactory;
+import org.apache.beam.runners.fnexecution.control.JobBundleFactory;
+import org.apache.beam.runners.fnexecution.control.StageBundleFactory;
+import org.apache.beam.runners.fnexecution.provisioning.JobInfo;
+import org.apache.beam.runners.fnexecution.state.StateRequestHandler;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Implementation of a {@link FlinkExecutableStageContext} for batch jobs. */
+class BatchFlinkExecutableStageContext implements FlinkExecutableStageContext {
+ private static final Logger LOG =
LoggerFactory.getLogger(BatchFlinkExecutableStageContext.class);
+
+ private final JobBundleFactory jobBundleFactory;
+ private final ArtifactSourcePool artifactSourcePool;
+
+ private static BatchFlinkExecutableStageContext create(JobInfo jobInfo)
throws Exception {
+ ArtifactSourcePool artifactSourcePool = ArtifactSourcePool.create();
+ JobBundleFactory jobBundleFactory = DockerJobBundleFactory.create(jobInfo,
artifactSourcePool);
+ return new BatchFlinkExecutableStageContext(jobBundleFactory,
artifactSourcePool);
+ }
+
+ private BatchFlinkExecutableStageContext(
+ JobBundleFactory jobBundleFactory, ArtifactSourcePool
artifactSourcePool) {
+ this.jobBundleFactory = jobBundleFactory;
+ this.artifactSourcePool = artifactSourcePool;
+ }
+
+ @Override
+ public <InputT> StageBundleFactory getStageBundleFactory(ExecutableStage
executableStage) {
+ return jobBundleFactory.<InputT>forStage(executableStage);
+ }
+
+ @Override
+ public StateRequestHandler getStateRequestHandler(
+ ExecutableStage executableStage, RuntimeContext runtimeContext) {
+ return FlinkBatchStateRequestHandler.forStage(executableStage,
runtimeContext);
+ }
+
+ @Override
+ public ArtifactSourcePool getArtifactSourcePool() {
+ return artifactSourcePool;
+ }
+
+ private void cleanUp() throws Exception {
+ jobBundleFactory.close();
+ }
+
+ enum BatchFactory implements Factory {
+ INSTANCE;
+
+ private final LoadingCache<JobInfo, BatchFlinkExecutableStageContext>
cachedContexts;
+
+ BatchFactory() {
+ cachedContexts =
+ CacheBuilder.newBuilder()
+ .weakValues()
+ .removalListener(
+ (RemovalNotification<JobInfo,
BatchFlinkExecutableStageContext> removal) -> {
+ try {
+ removal.getValue().cleanUp();
+ } catch (Exception e) {
+ LOG.warn(
+ "Error cleaning up bundle factory for job " +
removal.getKey().jobId(),
+ e);
+ }
+ })
+ .build(
+ new CacheLoader<JobInfo, BatchFlinkExecutableStageContext>()
{
Review comment:
Can this be a lambda too?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 110196)
Time Spent: 40m (was: 0.5h)
> Implement Flink batch ExecutableStage context
> ---------------------------------------------
>
> Key: BEAM-4523
> URL: https://issues.apache.org/jira/browse/BEAM-4523
> Project: Beam
> Issue Type: Bug
> Components: runner-flink
> Reporter: Ben Sidhom
> Assignee: Ben Sidhom
> Priority: Major
> Time Spent: 40m
> Remaining Estimate: 0h
>
> The the ExecutableStage context is a wrapper for the job and stage bundle
> factories and pooled artifact sources. It should take care of caching the
> overall job bundle factory since we do not have access to job lifecycle hooks
> in Flink but would like to reuse services and resources across operators
> within a given job.
> FlinkExecutableStageContext already exists as a skeleton, but it needs to be
> fleshed out.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
