xinyuiscool commented on code in PR #22576:
URL: https://github.com/apache/beam/pull/22576#discussion_r951731019
##########
runners/samza/src/main/java/org/apache/beam/runners/samza/runtime/OpAdapter.java:
##########
@@ -51,20 +54,23 @@
private static final Logger LOG = LoggerFactory.getLogger(OpAdapter.class);
private final Op<InT, OutT, K> op;
+ private final String opName;
private transient List<OpMessage<OutT>> outputList;
private transient CompletionStage<Collection<OpMessage<OutT>>> outputFuture;
private transient Instant outputWatermark;
private transient OpEmitter<OutT> emitter;
private transient Config config;
private transient Context context;
+ private transient List<SamzaPipelineExceptionListener.Registrar>
exceptionListeners;
public static <InT, OutT, K> AsyncFlatMapFunction<OpMessage<InT>,
OpMessage<OutT>> adapt(
- Op<InT, OutT, K> op) {
- return new OpAdapter<>(op);
+ Op<InT, OutT, K> op, String opName) {
Review Comment:
Instead of opName, please pass in translation context here directly. In the
future we might need to pull in more stuff from context to create the adapter,
and I don't want to change this factory method everytime we need a new field.
The changes on the constructor is right, and since it's private, we can make
local changes to add more fields.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
