[
https://issues.apache.org/jira/browse/BEAM-3977?focusedWorklogId=102333&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-102333
]
ASF GitHub Bot logged work on BEAM-3977:
----------------------------------------
Author: ASF GitHub Bot
Created on: 15/May/18 22:58
Start Date: 15/May/18 22:58
Worklog Time Spent: 10m
Work Description: axelmagn closed pull request #4988: [BEAM-3977] Move
out nested classes from SdkHarnessClient.
URL: https://github.com/apache/beam/pull/4988
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ActiveBundle.java
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ActiveBundle.java
new file mode 100644
index 00000000000..1a7453cf8b8
--- /dev/null
+++
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/ActiveBundle.java
@@ -0,0 +1,119 @@
+package org.apache.beam.runners.fnexecution.control;
+
+import java.util.Map;
+import java.util.concurrent.CompletionStage;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.runners.fnexecution.state.StateDelegator;
+import org.apache.beam.sdk.fn.data.CloseableFnDataReceiver;
+import org.apache.beam.sdk.fn.data.FnDataReceiver;
+import org.apache.beam.sdk.fn.data.InboundDataClient;
+import org.apache.beam.sdk.util.MoreFutures;
+import org.apache.beam.sdk.util.WindowedValue;
+
+/** An active bundle for a particular {@link
BeamFnApi.ProcessBundleDescriptor}. */
+public class ActiveBundle<InputT> implements AutoCloseable {
+ private final String bundleId;
+ private final CompletionStage<BeamFnApi.ProcessBundleResponse> response;
+ private final CloseableFnDataReceiver<WindowedValue<InputT>> inputReceiver;
+ private final Map<BeamFnApi.Target, InboundDataClient> outputClients;
+ private final StateDelegator.Registration stateRegistration;
+
+ ActiveBundle(
+ String bundleId,
+ CompletionStage<BeamFnApi.ProcessBundleResponse> response,
+ CloseableFnDataReceiver<WindowedValue<InputT>> inputReceiver,
+ Map<BeamFnApi.Target, InboundDataClient> outputClients,
+ StateDelegator.Registration stateRegistration) {
+ this.bundleId = bundleId;
+ this.response = response;
+ this.inputReceiver = inputReceiver;
+ this.outputClients = outputClients;
+ this.stateRegistration = stateRegistration;
+ }
+
+ /**
+ * Returns an id used to represent this bundle.
+ */
+ public String getBundleId() {
+ return bundleId;
+ }
+
+ /**
+ * Returns a {@link FnDataReceiver receiver} which consumes input elements
forwarding them
+ * to the SDK.
+ */
+ public FnDataReceiver<WindowedValue<InputT>> getInputReceiver() {
+ return inputReceiver;
+ }
+
+ /**
+ * Blocks till bundle processing is finished. This is comprised of:
+ * <ul>
+ * <li>closing the {@link #getInputReceiver() input receiver}.</li>
+ * <li>waiting for the SDK to say that processing the bundle is
finished.</li>
+ * <li>waiting for all inbound data clients to complete</li>
+ * </ul>
+ *
+ * <p>This method will throw an exception if bundle processing has failed.
+ * {@link Throwable#getSuppressed()} will return all the reasons as to why
processing has
+ * failed.
+ */
+ @Override
+ public void close() throws Exception {
+ Exception exception = null;
+ try {
+ inputReceiver.close();
+ } catch (Exception e) {
+ exception = e;
+ }
+ try {
+ // We don't have to worry about the completion stage.
+ if (exception == null) {
+ MoreFutures.get(response);
+ } else {
+ // TODO: [BEAM-3962] Handle aborting the bundle being processed.
+ throw new IllegalStateException("Processing bundle failed, "
+ + "TODO: [BEAM-3962] abort bundle.");
+ }
+ } catch (Exception e) {
+ if (exception == null) {
+ exception = e;
+ } else {
+ exception.addSuppressed(e);
+ }
+ }
+ try {
+ if (exception == null) {
+ stateRegistration.deregister();
+ } else {
+ stateRegistration.abort();
+ }
+ } catch (Exception e) {
+ if (exception == null) {
+ exception = e;
+ } else {
+ exception.addSuppressed(e);
+ }
+ }
+ for (InboundDataClient outputClient : outputClients.values()) {
+ try {
+ // If we failed processing this bundle, we should cancel all inbound
data.
+ if (exception == null) {
+ outputClient.awaitCompletion();
+ } else {
+ outputClient.cancel();
+ }
+ } catch (Exception e) {
+ if (exception == null) {
+ exception = e;
+ } else {
+ exception.addSuppressed(e);
+ }
+ }
+ }
+ if (exception != null) {
+ throw exception;
+ }
+
+ }
+}
diff --git
a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/BundleProcessor.java
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/BundleProcessor.java
new file mode 100644
index 00000000000..166f9bf79fd
--- /dev/null
+++
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/BundleProcessor.java
@@ -0,0 +1,151 @@
+package org.apache.beam.runners.fnexecution.control;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletionStage;
+import org.apache.beam.model.fnexecution.v1.BeamFnApi;
+import org.apache.beam.runners.fnexecution.data.FnDataService;
+import org.apache.beam.runners.fnexecution.data.RemoteInputDestination;
+import org.apache.beam.runners.fnexecution.state.StateDelegator;
+import org.apache.beam.runners.fnexecution.state.StateRequestHandler;
+import org.apache.beam.sdk.fn.data.CloseableFnDataReceiver;
+import org.apache.beam.sdk.fn.data.InboundDataClient;
+import org.apache.beam.sdk.fn.data.LogicalEndpoint;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A processor capable of creating bundles for some registered
+ * {@link BeamFnApi.ProcessBundleDescriptor}.
+ */
+public class BundleProcessor<T> {
+ private static final Logger LOG =
LoggerFactory.getLogger(BundleProcessor.class);
+
+ private final BeamFnApi.ProcessBundleDescriptor processBundleDescriptor;
+ private final CompletionStage<BeamFnApi.RegisterResponse> registrationFuture;
+ private final RemoteInputDestination<WindowedValue<T>> remoteInput;
+ private final StateDelegator stateDelegator;
+ private InstructionRequestHandler fnApiControlClient;
+ private FnDataService fnApiDataService;
+ private SdkHarnessClient.IdGenerator idGenerator;
+
+ BundleProcessor(
+ InstructionRequestHandler fnApiControlClient,
+ FnDataService fnApiDataService,
+ SdkHarnessClient.IdGenerator idGenerator,
+ BeamFnApi.ProcessBundleDescriptor processBundleDescriptor,
+ CompletionStage<BeamFnApi.RegisterResponse> registrationFuture,
+ RemoteInputDestination<WindowedValue<T>> remoteInput,
+ StateDelegator stateDelegator) {
+ this.processBundleDescriptor = processBundleDescriptor;
+ this.registrationFuture = registrationFuture;
+ this.remoteInput = remoteInput;
+ this.stateDelegator = stateDelegator;
+ this.fnApiControlClient = fnApiControlClient;
+ this.fnApiDataService = fnApiDataService;
+ this.idGenerator = idGenerator;
+ }
+
+ public CompletionStage<BeamFnApi.RegisterResponse> getRegistrationFuture() {
+ return registrationFuture;
+ }
+
+ /**
+ * Start a new bundle for the given {@link
BeamFnApi.ProcessBundleDescriptor} identifier.
+ *
+ * <p>The input channels for the returned {@link ActiveBundle} are derived
from the instructions
+ * in the {@link BeamFnApi.ProcessBundleDescriptor}.
+ *
+ * <p>NOTE: It is important to {@link ActiveBundle#close()} each bundle
after all elements are
+ * emitted.
+ *
+ * <pre>{@code
+ * try (ActiveBundle<InputT> bundle = SdkHarnessClient.newBundle(...)) {
+ * FnDataReceiver<InputT> inputReceiver = bundle.getInputReceiver();
+ * // send all elements ...
+ * }
+ * }</pre>
+ */
+ public ActiveBundle<T> newBundle(
+ Map<BeamFnApi.Target, RemoteOutputReceiver<?>> outputReceivers) {
+ return newBundle(outputReceivers, request -> {
+ throw new UnsupportedOperationException(String.format(
+ "The %s does not have a registered state handler.",
+ ActiveBundle.class.getSimpleName()));
+ });
+ }
+
+ /**
+ * Start a new bundle for the given {@link
BeamFnApi.ProcessBundleDescriptor} identifier.
+ *
+ * <p>The input channels for the returned {@link ActiveBundle} are derived
from the instructions
+ * in the {@link BeamFnApi.ProcessBundleDescriptor}.
+ *
+ * <p>NOTE: It is important to {@link ActiveBundle#close()} each bundle
after all elements are
+ * emitted.
+ *
+ * <pre>{@code
+ * try (ActiveBundle<InputT> bundle = SdkHarnessClient.newBundle(...)) {
+ * FnDataReceiver<InputT> inputReceiver = bundle.getInputReceiver();
+ * // send all elements ...
+ * }
+ * }</pre>
+ */
+ public ActiveBundle<T> newBundle(
+ Map<BeamFnApi.Target, RemoteOutputReceiver<?>> outputReceivers,
+ StateRequestHandler stateRequestHandler) {
+ String bundleId = idGenerator.getId();
+
+ final CompletionStage<BeamFnApi.InstructionResponse> genericResponse =
+ fnApiControlClient.handle(
+ BeamFnApi.InstructionRequest.newBuilder()
+ .setInstructionId(bundleId)
+ .setProcessBundle(
+ BeamFnApi.ProcessBundleRequest.newBuilder()
+
.setProcessBundleDescriptorReference(processBundleDescriptor.getId()))
+ .build());
+ LOG.debug(
+ "Sent {} with ID {} for {} with ID {}",
+ BeamFnApi.ProcessBundleRequest.class.getSimpleName(),
+ bundleId,
+ BeamFnApi.ProcessBundleDescriptor.class.getSimpleName(),
+ processBundleDescriptor.getId());
+
+ CompletionStage<BeamFnApi.ProcessBundleResponse> specificResponse =
+
genericResponse.thenApply(BeamFnApi.InstructionResponse::getProcessBundle);
+ Map<BeamFnApi.Target, InboundDataClient> outputClients = new HashMap<>();
+ for (Map.Entry<BeamFnApi.Target, RemoteOutputReceiver<?>> targetReceiver
+ : outputReceivers.entrySet()) {
+ InboundDataClient outputClient =
+ attachReceiver(
+ bundleId,
+ targetReceiver.getKey(),
+ (RemoteOutputReceiver) targetReceiver.getValue());
+ outputClients.put(targetReceiver.getKey(), outputClient);
+ }
+
+ CloseableFnDataReceiver<WindowedValue<T>> dataReceiver =
+ fnApiDataService.send(
+ LogicalEndpoint.of(bundleId, remoteInput.getTarget()),
remoteInput.getCoder());
+
+ return new ActiveBundle(
+ bundleId,
+ specificResponse,
+ dataReceiver,
+ outputClients,
+ stateDelegator.registerForProcessBundleInstructionId(bundleId,
stateRequestHandler));
+ }
+
+ private <OutputT> InboundDataClient attachReceiver(
+ String bundleId,
+ BeamFnApi.Target target,
+ RemoteOutputReceiver<WindowedValue<OutputT>> receiver) {
+ return fnApiDataService.receive(
+ LogicalEndpoint.of(bundleId, target), receiver.getCoder(),
receiver.getReceiver());
+ }
+
+ BeamFnApi.ProcessBundleDescriptor getProcessBundleDescriptor() {
+ return processBundleDescriptor;
+ }
+}
diff --git
a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/RemoteOutputReceiver.java
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/RemoteOutputReceiver.java
new file mode 100644
index 00000000000..59f69189330
--- /dev/null
+++
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/RemoteOutputReceiver.java
@@ -0,0 +1,20 @@
+package org.apache.beam.runners.fnexecution.control;
+
+import com.google.auto.value.AutoValue;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.fn.data.FnDataReceiver;
+import org.apache.beam.sdk.fn.data.LogicalEndpoint;
+
+/**
+ * A pair of {@link Coder} and {@link FnDataReceiver} which can be registered
to receive elements
+ * for a {@link LogicalEndpoint}.
+ */
+@AutoValue
+public abstract class RemoteOutputReceiver<T> {
+ public static <T> RemoteOutputReceiver of (Coder<T> coder, FnDataReceiver<T>
receiver) {
+ return new AutoValue_RemoteOutputReceiver<>(coder, receiver);
+ }
+
+ public abstract Coder<T> getCoder();
+ public abstract FnDataReceiver<T> getReceiver();
+}
diff --git
a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClient.java
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClient.java
index 0447cf4574c..35aef3b3df6 100644
---
a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClient.java
+++
b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClient.java
@@ -20,28 +20,19 @@
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
-import com.google.auto.value.AutoValue;
-import java.util.HashMap;
-import java.util.Map;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.beam.model.fnexecution.v1.BeamFnApi;
import org.apache.beam.model.fnexecution.v1.BeamFnApi.InstructionResponse;
import org.apache.beam.model.fnexecution.v1.BeamFnApi.ProcessBundleDescriptor;
-import org.apache.beam.model.fnexecution.v1.BeamFnApi.ProcessBundleRequest;
import org.apache.beam.model.fnexecution.v1.BeamFnApi.RegisterResponse;
import org.apache.beam.model.pipeline.v1.Endpoints;
import org.apache.beam.runners.fnexecution.data.FnDataService;
import org.apache.beam.runners.fnexecution.data.RemoteInputDestination;
import org.apache.beam.runners.fnexecution.state.StateDelegator;
import org.apache.beam.runners.fnexecution.state.StateRequestHandler;
-import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.fn.data.CloseableFnDataReceiver;
-import org.apache.beam.sdk.fn.data.FnDataReceiver;
-import org.apache.beam.sdk.fn.data.InboundDataClient;
-import org.apache.beam.sdk.fn.data.LogicalEndpoint;
-import org.apache.beam.sdk.util.MoreFutures;
import org.apache.beam.sdk.util.WindowedValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -74,225 +65,31 @@ public String getId() {
}
/**
- * A processor capable of creating bundles for some registered {@link
ProcessBundleDescriptor}.
+ * A {@link StateDelegator} that issues zero state requests to any provided
+ * {@link StateRequestHandler state handlers}.
*/
- public class BundleProcessor<T> {
- private final ProcessBundleDescriptor processBundleDescriptor;
- private final CompletionStage<RegisterResponse> registrationFuture;
- private final RemoteInputDestination<WindowedValue<T>> remoteInput;
- private final StateDelegator stateDelegator;
-
- private BundleProcessor(
- ProcessBundleDescriptor processBundleDescriptor,
- CompletionStage<RegisterResponse> registrationFuture,
- RemoteInputDestination<WindowedValue<T>> remoteInput,
- StateDelegator stateDelegator) {
- this.processBundleDescriptor = processBundleDescriptor;
- this.registrationFuture = registrationFuture;
- this.remoteInput = remoteInput;
- this.stateDelegator = stateDelegator;
- }
-
- public CompletionStage<RegisterResponse> getRegistrationFuture() {
- return registrationFuture;
- }
-
- /**
- * Start a new bundle for the given {@link
BeamFnApi.ProcessBundleDescriptor} identifier.
- *
- * <p>The input channels for the returned {@link ActiveBundle} are derived
from the instructions
- * in the {@link BeamFnApi.ProcessBundleDescriptor}.
- *
- * <p>NOTE: It is important to {@link #close()} each bundle after all
elements are emitted.
- * <pre>{@code
- * try (ActiveBundle<InputT> bundle = SdkHarnessClient.newBundle(...)) {
- * FnDataReceiver<InputT> inputReceiver = bundle.getInputReceiver();
- * // send all elements ...
- * }
- * }</pre>
- */
- public ActiveBundle<T> newBundle(
- Map<BeamFnApi.Target, RemoteOutputReceiver<?>> outputReceivers) {
- return newBundle(outputReceivers, request -> {
- throw new UnsupportedOperationException(String.format(
- "The %s does not have a registered state handler.",
- ActiveBundle.class.getSimpleName()));
- });
+ private static class NoOpStateDelegator implements StateDelegator {
+ private static final NoOpStateDelegator INSTANCE = new
NoOpStateDelegator();
+ @Override
+ public Registration registerForProcessBundleInstructionId(
+ String processBundleInstructionId,
+ StateRequestHandler handler) {
+ return Registration.INSTANCE;
}
/**
- * Start a new bundle for the given {@link
BeamFnApi.ProcessBundleDescriptor} identifier.
- *
- * <p>The input channels for the returned {@link ActiveBundle} are derived
from the instructions
- * in the {@link BeamFnApi.ProcessBundleDescriptor}.
- *
- * <p>NOTE: It is important to {@link #close()} each bundle after all
elements are emitted.
- * <pre>{@code
- * try (ActiveBundle<InputT> bundle = SdkHarnessClient.newBundle(...)) {
- * FnDataReceiver<InputT> inputReceiver = bundle.getInputReceiver();
- * // send all elements ...
- * }
- * }</pre>
+ * The corresponding registration for a {@link NoOpStateDelegator} that
does nothing.
*/
- public ActiveBundle<T> newBundle(
- Map<BeamFnApi.Target, RemoteOutputReceiver<?>> outputReceivers,
- StateRequestHandler stateRequestHandler) {
- String bundleId = idGenerator.getId();
-
- final CompletionStage<BeamFnApi.InstructionResponse> genericResponse =
- fnApiControlClient.handle(
- BeamFnApi.InstructionRequest.newBuilder()
- .setInstructionId(bundleId)
- .setProcessBundle(
- BeamFnApi.ProcessBundleRequest.newBuilder()
-
.setProcessBundleDescriptorReference(processBundleDescriptor.getId()))
- .build());
- LOG.debug(
- "Sent {} with ID {} for {} with ID {}",
- ProcessBundleRequest.class.getSimpleName(),
- bundleId,
- ProcessBundleDescriptor.class.getSimpleName(),
- processBundleDescriptor.getId());
+ private static class Registration implements StateDelegator.Registration {
+ private static final Registration INSTANCE = new Registration();
- CompletionStage<BeamFnApi.ProcessBundleResponse> specificResponse =
- genericResponse.thenApply(InstructionResponse::getProcessBundle);
- Map<BeamFnApi.Target, InboundDataClient> outputClients = new HashMap<>();
- for (Map.Entry<BeamFnApi.Target, RemoteOutputReceiver<?>> targetReceiver
- : outputReceivers.entrySet()) {
- InboundDataClient outputClient =
- attachReceiver(
- bundleId,
- targetReceiver.getKey(),
- (RemoteOutputReceiver) targetReceiver.getValue());
- outputClients.put(targetReceiver.getKey(), outputClient);
+ @Override
+ public void deregister() {
}
- CloseableFnDataReceiver<WindowedValue<T>> dataReceiver =
- fnApiDataService.send(
- LogicalEndpoint.of(bundleId, remoteInput.getTarget()),
remoteInput.getCoder());
-
- return new ActiveBundle(
- bundleId,
- specificResponse,
- dataReceiver,
- outputClients,
- stateDelegator.registerForProcessBundleInstructionId(bundleId,
stateRequestHandler));
- }
-
- private <OutputT> InboundDataClient attachReceiver(
- String bundleId,
- BeamFnApi.Target target,
- RemoteOutputReceiver<WindowedValue<OutputT>> receiver) {
- return fnApiDataService.receive(
- LogicalEndpoint.of(bundleId, target), receiver.getCoder(),
receiver.getReceiver());
- }
- }
-
- /** An active bundle for a particular {@link
BeamFnApi.ProcessBundleDescriptor}. */
- public static class ActiveBundle<InputT> implements AutoCloseable {
- private final String bundleId;
- private final CompletionStage<BeamFnApi.ProcessBundleResponse> response;
- private final CloseableFnDataReceiver<WindowedValue<InputT>> inputReceiver;
- private final Map<BeamFnApi.Target, InboundDataClient> outputClients;
- private final StateDelegator.Registration stateRegistration;
-
- private ActiveBundle(
- String bundleId,
- CompletionStage<BeamFnApi.ProcessBundleResponse> response,
- CloseableFnDataReceiver<WindowedValue<InputT>> inputReceiver,
- Map<BeamFnApi.Target, InboundDataClient> outputClients,
- StateDelegator.Registration stateRegistration) {
- this.bundleId = bundleId;
- this.response = response;
- this.inputReceiver = inputReceiver;
- this.outputClients = outputClients;
- this.stateRegistration = stateRegistration;
- }
-
- /**
- * Returns an id used to represent this bundle.
- */
- public String getBundleId() {
- return bundleId;
- }
-
- /**
- * Returns a {@link FnDataReceiver receiver} which consumes input elements
forwarding them
- * to the SDK.
- */
- public FnDataReceiver<WindowedValue<InputT>> getInputReceiver() {
- return inputReceiver;
- }
-
- /**
- * Blocks till bundle processing is finished. This is comprised of:
- * <ul>
- * <li>closing the {@link #getInputReceiver() input receiver}.</li>
- * <li>waiting for the SDK to say that processing the bundle is
finished.</li>
- * <li>waiting for all inbound data clients to complete</li>
- * </ul>
- *
- * <p>This method will throw an exception if bundle processing has failed.
- * {@link Throwable#getSuppressed()} will return all the reasons as to why
processing has
- * failed.
- */
- @Override
- public void close() throws Exception {
- Exception exception = null;
- try {
- inputReceiver.close();
- } catch (Exception e) {
- exception = e;
- }
- try {
- // We don't have to worry about the completion stage.
- if (exception == null) {
- MoreFutures.get(response);
- } else {
- // TODO: [BEAM-3962] Handle aborting the bundle being processed.
- throw new IllegalStateException("Processing bundle failed, "
- + "TODO: [BEAM-3962] abort bundle.");
- }
- } catch (Exception e) {
- if (exception == null) {
- exception = e;
- } else {
- exception.addSuppressed(e);
- }
- }
- try {
- if (exception == null) {
- stateRegistration.deregister();
- } else {
- stateRegistration.abort();
- }
- } catch (Exception e) {
- if (exception == null) {
- exception = e;
- } else {
- exception.addSuppressed(e);
- }
- }
- for (InboundDataClient outputClient : outputClients.values()) {
- try {
- // If we failed processing this bundle, we should cancel all inbound
data.
- if (exception == null) {
- outputClient.awaitCompletion();
- } else {
- outputClient.cancel();
- }
- } catch (Exception e) {
- if (exception == null) {
- exception = e;
- } else {
- exception.addSuppressed(e);
- }
- }
- }
- if (exception != null) {
- throw exception;
+ @Override
+ public void abort() {
}
-
}
}
@@ -351,33 +148,6 @@ public SdkHarnessClient withIdGenerator(IdGenerator
idGenerator) {
return getProcessor(descriptor, remoteInputDesination,
NoOpStateDelegator.INSTANCE);
}
- /**
- * A {@link StateDelegator} that issues zero state requests to any provided
- * {@link StateRequestHandler state handlers}.
- */
- private static class NoOpStateDelegator implements StateDelegator {
- private static final NoOpStateDelegator INSTANCE = new
NoOpStateDelegator();
- @Override
- public Registration registerForProcessBundleInstructionId(String
processBundleInstructionId,
- StateRequestHandler handler) {
- return Registration.INSTANCE;
- }
-
- /**
- * The corresponding registration for a {@link NoOpStateDelegator} that
does nothing.
- */
- private static class Registration implements StateDelegator.Registration {
- private static final Registration INSTANCE = new Registration();
-
- @Override
- public void deregister() {
- }
-
- @Override
- public void abort() {
- }
- }
- }
/**
* Provides {@link BundleProcessor} that is capable of processing bundles
containing
@@ -394,16 +164,16 @@ public void abort() {
*/
public <T> BundleProcessor<T> getProcessor(
BeamFnApi.ProcessBundleDescriptor descriptor,
- RemoteInputDestination<WindowedValue<T>> remoteInputDesination,
+ RemoteInputDestination<WindowedValue<T>> remoteInputDestination,
StateDelegator stateDelegator) {
BundleProcessor<T> bundleProcessor =
clientProcessors.computeIfAbsent(
descriptor.getId(),
s -> create(
descriptor,
- (RemoteInputDestination) remoteInputDesination,
+ (RemoteInputDestination) remoteInputDestination,
stateDelegator));
- checkArgument(bundleProcessor.processBundleDescriptor.equals(descriptor),
+
checkArgument(bundleProcessor.getProcessBundleDescriptor().equals(descriptor),
"The provided %s with id %s collides with an existing %s with the same
id but "
+ "containing different contents.",
BeamFnApi.ProcessBundleDescriptor.class.getSimpleName(),
@@ -436,7 +206,7 @@ public void abort() {
genericResponse.thenApply(InstructionResponse::getRegister);
BundleProcessor<T> bundleProcessor = new BundleProcessor<>(
- processBundleDescriptor,
+ fnApiControlClient, fnApiDataService, idGenerator,
processBundleDescriptor,
registerResponseFuture,
remoteInputDestination,
stateDelegator);
@@ -447,17 +217,4 @@ public void abort() {
@Override
public void close() {}
- /**
- * A pair of {@link Coder} and {@link FnDataReceiver} which can be
registered to receive elements
- * for a {@link LogicalEndpoint}.
- */
- @AutoValue
- public abstract static class RemoteOutputReceiver<T> {
- public static <T> RemoteOutputReceiver of (Coder<T> coder,
FnDataReceiver<T> receiver) {
- return new AutoValue_SdkHarnessClient_RemoteOutputReceiver<>(coder,
receiver);
- }
-
- public abstract Coder<T> getCoder();
- public abstract FnDataReceiver<T> getReceiver();
- }
}
diff --git
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/RemoteExecutionTest.java
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/RemoteExecutionTest.java
index 962cb790b27..88ab8cf87ef 100644
---
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/RemoteExecutionTest.java
+++
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/RemoteExecutionTest.java
@@ -50,9 +50,6 @@
import org.apache.beam.runners.fnexecution.GrpcFnServer;
import org.apache.beam.runners.fnexecution.InProcessServerFactory;
import
org.apache.beam.runners.fnexecution.control.ProcessBundleDescriptors.ExecutableProcessBundleDescriptor;
-import
org.apache.beam.runners.fnexecution.control.SdkHarnessClient.ActiveBundle;
-import
org.apache.beam.runners.fnexecution.control.SdkHarnessClient.BundleProcessor;
-import
org.apache.beam.runners.fnexecution.control.SdkHarnessClient.RemoteOutputReceiver;
import org.apache.beam.runners.fnexecution.data.GrpcDataService;
import org.apache.beam.runners.fnexecution.data.RemoteInputDestination;
import org.apache.beam.runners.fnexecution.logging.GrpcLoggingService;
diff --git
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClientTest.java
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClientTest.java
index 1a229bfda5d..45f7a6a1238 100644
---
a/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClientTest.java
+++
b/runners/java-fn-execution/src/test/java/org/apache/beam/runners/fnexecution/control/SdkHarnessClientTest.java
@@ -51,9 +51,6 @@
import org.apache.beam.runners.core.construction.CoderTranslation;
import org.apache.beam.runners.core.construction.PipelineTranslation;
import org.apache.beam.runners.fnexecution.InProcessSdkHarness;
-import
org.apache.beam.runners.fnexecution.control.SdkHarnessClient.ActiveBundle;
-import
org.apache.beam.runners.fnexecution.control.SdkHarnessClient.BundleProcessor;
-import
org.apache.beam.runners.fnexecution.control.SdkHarnessClient.RemoteOutputReceiver;
import org.apache.beam.runners.fnexecution.data.FnDataService;
import org.apache.beam.runners.fnexecution.data.RemoteInputDestination;
import org.apache.beam.runners.fnexecution.state.StateDelegator;
----------------------------------------------------------------
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: 102333)
Time Spent: 2h 10m (was: 2h)
> Member classes of SdkHarnessClient should have their own files.
> ---------------------------------------------------------------
>
> Key: BEAM-3977
> URL: https://issues.apache.org/jira/browse/BEAM-3977
> Project: Beam
> Issue Type: Improvement
> Components: runner-core
> Reporter: Axel Magnuson
> Assignee: Axel Magnuson
> Priority: Minor
> Time Spent: 2h 10m
> Remaining Estimate: 0h
>
> SdkHarnessClient contains quite a few nested classes that could be split out.
> of these, BundleProcessor and ActiveBundle have grown up to be first class
> concepts that we interact with just as much as the SdkHarnessClient.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
