Upgrade transforms without upgrading the pipelines #28210

chamikaramj · 2023-08-29T21:04:07Z

This uses the Beam Transform Service to support upgrading Beam Java transforms used by pipelines without upgrading the Beam version used by the pipelines.

A similar solution can be introduced for other SDKs in the future.

For more details please see #27943.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

robertwb · 2023-08-31T15:34:54Z

...ruction-java/src/main/java/org/apache/beam/runners/core/construction/CombineTranslation.java

@@ -62,6 +64,11 @@ private CombinePerKeyPayloadTranslator() {}

    @Override
    public String getUrn(Combine.PerKey<?, ?, ?> transform) {
+      return getUrn();


Should this be in the base class?

Yeah, moved.

robertwb · 2023-08-31T15:37:20Z

...ruction-java/src/main/java/org/apache/beam/runners/core/construction/CombineTranslation.java

@@ -84,6 +91,16 @@ public FunctionSpec translate(
      }
    }

+    @Override
+    public @Nullable Row toConfigRow(PTransform<?, ?> pTransform) {


I wonder if these should be in an optional interface, or if default null-returning implementations should be provided. (We could argue that it should be strongly encouraged.)

I don't think these need a separate interface but good point regarding providing default implementations. Made "toConfigRow" and "fromConfigRow" default methods of the interface so that sub-classes can choose to not implement them.

robertwb · 2023-08-31T15:40:49Z

...ruction-java/src/main/java/org/apache/beam/runners/core/construction/ImpulseTranslation.java

+    @Override
+    public @Nullable Row toConfigRow(PTransform<?, ?> pTransform) {
+      Impulse impulse = (Impulse) pTransform;
+      System.out.println("Found impulse transform: " + impulse);


Remove debugging.

robertwb · 2023-08-31T15:45:37Z

...tion-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

+        }
+
+        ExpandableTransform externalTransform =
+            External.of(urn, payload.toByteArray(), serviceAddress);


serviceAddress might be unset. (Perhaps don't set it to null above to let the compiler check this for you.)

Added a check to confirm that it's not null (either should be provided by user or should be the address of the Transform Service we start here).

(Also this logic was moved to PipelineTranslator).

robertwb · 2023-08-31T15:47:03Z

...tion-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

+        }
+        externalTransform.expand(input);
+
+        return externalTransform.getExpandedTransform().getSpec();


How do the outputs get wired up correctly?

Added logic to wire inputs and outputs correctly (in PipelineTranslator now).

.../expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java

robertwb · 2023-08-31T15:56:21Z

.../expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java

+            continue;
+          }
+
+          String urn = null;


I think we can leave this unset and the compiler is smart enough to see the assignment or continue.

This code was reverted.

I'm still seeing it.

Should be done now.

robertwb · 2023-08-31T15:57:55Z

.../expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java

+          try {
+            urn = translator.getUrn();
+            if (urn == null) {
+              LOG.info(


Maybe the warning should be softer, that this translator does not have a tranform-independent URN?

Yeah, changed to debug and updated.

robertwb · 2023-08-31T15:58:56Z

.../expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java

+                      + " to the Expansion Service.");
+              continue;
+            }
+          } catch (Exception e) {


Why would this happen? Feels that this is something exceptional enough to propagate up.

This is general catch all to prevent a single incorrectly implemented PayloadTranslator available in the system from crashing to whole service. I think it's safer to catch such exceptions and continue to load the rest of the transforms (we do the same for Schema-aware transforms). Note that some of these implementations may come from custom/user libraries.

.../expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java

chamikaramj

Thanks Robert. PTAL.

chamikaramj · 2023-09-12T23:39:31Z

...ruction-java/src/main/java/org/apache/beam/runners/core/construction/ImpulseTranslation.java

+    @Override
+    public @Nullable Row toConfigRow(PTransform<?, ?> pTransform) {
+      Impulse impulse = (Impulse) pTransform;
+      System.out.println("Found impulse transform: " + impulse);


chamikaramj · 2023-09-12T23:48:05Z

...tion-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

+        }
+
+        ExpandableTransform externalTransform =
+            External.of(urn, payload.toByteArray(), serviceAddress);


Added a check to confirm that it's not null (either should be provided by user or should be the address of the Transform Service we start here).

chamikaramj · 2023-09-12T23:48:24Z

...tion-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

+        }
+
+        ExpandableTransform externalTransform =
+            External.of(urn, payload.toByteArray(), serviceAddress);


(Also this logic was moved to PipelineTranslator).

chamikaramj · 2023-09-12T23:49:50Z

...tion-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

+        }
+        externalTransform.expand(input);
+
+        return externalTransform.getExpandedTransform().getSpec();


Added logic to wire inputs and outputs correctly (in PipelineTranslator now).

chamikaramj · 2023-09-12T23:51:05Z

...tion-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

+      PTransform<?, ?> transform = appliedPTransform.getTransform();
+
+      FunctionSpec spec = null;
+      if (getUrn(transform) != null && urnsToOverride.contains(getUrn(transform))) {


I think translation time is a good fit for this since we are modifying the translated pipeline, but you are right that this has to be done at a higher level since otherwise it's impossible to wire inputs and outputs correctly.

I moved this logic to PipelineTranslator and fixed the wiring.

chamikaramj · 2023-09-13T00:00:21Z

.../expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java

+          try {
+            urn = translator.getUrn();
+            if (urn == null) {
+              LOG.info(


Yeah, changed to debug and updated.

chamikaramj · 2023-09-13T00:02:55Z

.../expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java

+                      + " to the Expansion Service.");
+              continue;
+            }
+          } catch (Exception e) {


This is general catch all to prevent a single incorrectly implemented PayloadTranslator available in the system from crashing to whole service. I think it's safer to catch such exceptions and continue to load the rest of the transforms (we do the same for Schema-aware transforms). Note that some of these implementations may come from custom/user libraries.

chamikaramj · 2023-09-13T04:08:03Z

...ruction-java/src/main/java/org/apache/beam/runners/core/construction/CombineTranslation.java

@@ -62,6 +64,11 @@ private CombinePerKeyPayloadTranslator() {}

    @Override
    public String getUrn(Combine.PerKey<?, ?, ?> transform) {
+      return getUrn();


Yeah, moved.

chamikaramj · 2023-09-13T04:09:44Z

...ruction-java/src/main/java/org/apache/beam/runners/core/construction/CombineTranslation.java

@@ -84,6 +91,16 @@ public FunctionSpec translate(
      }
    }

+    @Override
+    public @Nullable Row toConfigRow(PTransform<?, ?> pTransform) {


I don't think these need a separate interface but good point regarding providing default implementations. Made "toConfigRow" and "fromConfigRow" default methods of the interface so that sub-classes can choose to not implement them.

.../expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java

chamikaramj · 2023-09-13T04:34:10Z

Also confirmed that this work for Dataflow by running a pipeline that upgrades the Create transform to a version available in an expansion service.

https://console.cloud.google.com/dataflow/jobs/us-central1/2023-09-12_15_32_47-10327070264136367924?project=apache-beam-testing

cc: @kennknowles

github-actions · 2023-09-13T06:06:41Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

robertwb · 2023-09-15T21:32:00Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/CreateTranslation.java

+  }
+
+  @Override
+  public Create.@Nullable Values<?> fromConfigRow(Row configRow) {


Stray @Nullable (I don't think this should be nullable.)

This class was reverted.

robertwb · 2023-09-15T21:32:28Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/CreateTranslation.java

+  }
+
+  @Override
+  public @Nullable Row toConfigRow(Values<?> transform) {


Shouldn't be nullable.

This class was reverted.

robertwb · 2023-09-15T21:34:08Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/CreateTranslation.java

+import org.checkerframework.checker.nullness.qual.Nullable;
+
+@SuppressWarnings({
+  "rawtypes", // TODO(https://github.com/apache/beam/issues/20447)


Can we avoid suppressing rawTypes (or at least make it more local if it can't be avoided)?

Removed the suppression.

robertwb · 2023-09-15T21:36:00Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/CreateTranslation.java

+        .forEach(
+            object -> {
+              encodedElements.add(toByteArray(object));
+            });


If you're going to be using the new functional API, maybe use streams + collect to list?

This class was reverted.

robertwb · 2023-09-15T21:38:24Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/CreateTranslation.java

+            configRow.getArray("values").stream()
+                .map(bytesValue -> fromByteArray((byte[]) bytesValue))
+                .collect(Collectors.toList()));
+    byte[] serializedCoder = configRow.getBytes("serialized_coder");


Would coder ever be null? Preferably we simply require it.

This class was reverted.

robertwb · 2023-09-15T22:28:57Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+          SdkComponents.create(
+              runnerAPIpipeline.getComponents(), runnerAPIpipeline.getRequirementsList());
+      String updatedTransformId =
+          updatedComponents.registerPTransform(updatedAppliedPTransform, Collections.emptyList());


Collections.emptyList() doesn't seem correct if this is a composite transform.

robertwb · 2023-09-15T22:36:31Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+      // old one.
+      // Also recording the newly generated id of the old (overridden) transform in the
+      // updatedRunnerApiComponents.
+      Map<String, Map<String, String>> transformInputUpdates = new HashMap<>();


Could you document what transformInputUpdates contains? String isn't very descriptive.

robertwb · 2023-09-15T22:38:10Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+                + oldTransformIds.size()
+                + " were updated.");
+      }
+      String oldTransformId = oldTransformIds.get(0);


Shouldn't we already know exactly what transform we're updating? This seems a round-about way of getting it, and will break if a transform doesn't have outputs.

robertwb · 2023-09-15T22:38:42Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+      updaterRootTransformIds.addAll(runnerAPIpipeline.getRootTransformIdsList());
+      if (updaterRootTransformIds.contains(oldTransformId)) {
+        updaterRootTransformIds.remove(oldTransformId);
+        updaterRootTransformIds.add(updatedTransformId);


I suppose the order does not matter here?

Updated logic should preserve the order.

robertwb · 2023-09-15T22:47:50Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+      // Will contain the outputs of the upgraded transform.
+      Map<TupleTag<?>, PCollection<?>> newOutputs = new HashMap<>();
+
+      if (output instanceof PCollectionTuple) {


This code deals with both the Pipeline object and the Pipeline proto, and it's a bit hard to follow where the source of truth is. My personal preference would be to simply act on the protos consuming the expansion service response directly. Honestly, I don't think that'd be any more complicated than what you're already having to do here already.

Makes sense. Updated code (now in TransformUpgrader) to directly update the proto and directly invoke the TransformService using protos instead of going through External.java.

chamikaramj

Thanks Robert. PTAL.

chamikaramj · 2023-09-20T22:45:34Z

.../expansion-service/src/main/java/org/apache/beam/sdk/expansion/service/ExpansionService.java

+            continue;
+          }
+
+          String urn = null;


Should be done now.

chamikaramj · 2023-09-20T22:46:01Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/CreateTranslation.java

+import org.checkerframework.checker.nullness.qual.Nullable;
+
+@SuppressWarnings({
+  "rawtypes", // TODO(https://github.com/apache/beam/issues/20447)


Removed the suppression.

chamikaramj · 2023-09-20T22:46:15Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/CreateTranslation.java

+  }
+
+  @Override
+  public @Nullable Row toConfigRow(Values<?> transform) {


This class was reverted.

chamikaramj · 2023-09-20T22:46:19Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/CreateTranslation.java

+        .forEach(
+            object -> {
+              encodedElements.add(toByteArray(object));
+            });


This class was reverted.

chamikaramj · 2023-09-20T22:46:23Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/CreateTranslation.java

+              encodedElements.add(toByteArray(object));
+            });
+
+    byte[] serializedCoder =


This class was reverted.

chamikaramj · 2023-09-20T22:50:23Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+      // old one.
+      // Also recording the newly generated id of the old (overridden) transform in the
+      // updatedRunnerApiComponents.
+      Map<String, Map<String, String>> transformInputUpdates = new HashMap<>();


chamikaramj · 2023-09-20T22:50:30Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+                + oldTransformIds.size()
+                + " were updated.");
+      }
+      String oldTransformId = oldTransformIds.get(0);


chamikaramj · 2023-09-20T22:52:37Z

...uction-java/src/main/java/org/apache/beam/runners/core/construction/PipelineTranslation.java

+      updaterRootTransformIds.addAll(runnerAPIpipeline.getRootTransformIdsList());
+      if (updaterRootTransformIds.contains(oldTransformId)) {
+        updaterRootTransformIds.remove(oldTransformId);
+        updaterRootTransformIds.add(updatedTransformId);


Updated logic should preserve the order.

chamikaramj · 2023-09-20T22:57:16Z

...oogle-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java

@@ -2572,6 +2572,11 @@ public String getUrn(PTransform transform) {
      return "dataflow_stub:" + transform.getClass().getName();
    }

+    @Override
+    public String getUrn() {


The access modifier of interface method is public and Java doesn't allow reducing access in sub-classes.

With the updated interface, the callers should be able to directly invoke getUrn() unless the sub-class cannot determine the URN without looking at the transform (which should be rare and we should not use the same translator for multiple transform classes.

chamikaramj · 2023-09-20T23:04:41Z

...tion-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

+    /**
+     * Translates the given transform represented by the provided {@code AppliedPTransform} to a
+     * {@code FunctionSpec} with a URN and a payload.
+     */
    @Nullable


robertwb

Thanks, this is looking pretty good. Only some minor comments.

robertwb · 2023-09-21T21:01:55Z

...tion-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

+        // optional.
+      }
+      if (configRow != null) {
+        ByteStringOutputStream rowOutputStream = new ByteStringOutputStream();


I think there's a CoderUtils that does this for you.

robertwb · 2023-09-21T21:09:32Z

...tion-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

+
+        ByteStringOutputStream schemaOutputStream = new ByteStringOutputStream();
+        try (ObjectOutputStream schemaObjOut = new ObjectOutputStream(schemaOutputStream)) {
+          schemaObjOut.writeObject(SchemaTranslation.schemaToProto(configRow.getSchema(), true));


SchemaTranslation.schemaToProto is a Proto object. Just use toByteArray to serialize this proto. You can use ByteString.copyFrom to go from byte[] to ByteString and not have to deal with streams at all.

robertwb · 2023-09-21T21:14:40Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+      RunnerApi.Pipeline updateTransformViaTransformService(
+          RunnerApi.Pipeline runnerAPIpipeline,
+          String transformId,
+          Endpoints.ApiServiceDescriptor transformServiceEndpoint)


Perhaps create the client once and pass it here, rather than passing the endpoint and re-connecting it every time?

"DefaultExpansionServiceClientFactory" caches channels for endpoints. So I don't think this is an issue.

robertwb · 2023-09-21T21:17:32Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+   */
+  public RunnerApi.Pipeline upgradeTransformsViaTransformService(
+      RunnerApi.Pipeline pipeline, List<String> urnsToOverride, ExternalTranslationOptions options)
+      throws Exception {


Avoid throwing generic Exception. (And below.)

robertwb · 2023-09-21T21:21:08Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+    PTransform transformToUpgrade =
+        runnerAPIpipeline.getComponents().getTransformsMap().get(transformId);
+    if (transformToUpgrade == null) {
+      throw new Exception("Could not find a transform with the ID " + transformId);


Seems like a good candidate for Preconditions.checkNotNull(). Or throw something like IllegalArgumentException.

Updated to throw IllegalArgumentException.

robertwb · 2023-09-21T21:22:29Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+      throw new RuntimeException(e);
+    }
+
+    Row configRow =


You already have configRowBytes, no need to encode it just to recode it.

robertwb · 2023-09-21T21:24:54Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+        clientFactory.getExpansionServiceClient(transformServiceEndpoint).expand(request);
+
+    if (!Strings.isNullOrEmpty(response.getError())) {
+      throw new IOException(String.format("expansion service error: %s", response.getError()));


IOException doesn't seem to be the right thing to throw for a mis-configured transform.

Updated to RuntimeException.

robertwb · 2023-09-21T21:27:34Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+      }
+    }
+
+    String newTransformId = transformId + "_upgraded";


We can't guarantee this won't be a collision. On the other hand, there's no need to give it a new name at all, just put the new transform in the map at the old transforms name (and all references will remain valid).

robertwb · 2023-09-21T21:29:40Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+                        transformBuilder.putAllInputs(updatedInputsMap);
+                      }
+
+                      // Fix sub-transforms


If you keep the name, you won't have to do this.

robertwb · 2023-09-21T21:29:59Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+    newComponentsBuilder.putAllTransforms(updatedExpandedTransformMap);
+    newComponentsBuilder.putTransforms(newTransformId, expandedTransform);
+
+    // We fix the root in case the overridden transform was one of the roots.


chamikaramj

Thanks. PTAL.

chamikaramj · 2023-09-21T22:16:34Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+   */
+  public RunnerApi.Pipeline upgradeTransformsViaTransformService(
+      RunnerApi.Pipeline pipeline, List<String> urnsToOverride, ExternalTranslationOptions options)
+      throws Exception {


chamikaramj · 2023-09-21T22:20:05Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+      RunnerApi.Pipeline updateTransformViaTransformService(
+          RunnerApi.Pipeline runnerAPIpipeline,
+          String transformId,
+          Endpoints.ApiServiceDescriptor transformServiceEndpoint)


"DefaultExpansionServiceClientFactory" caches channels for endpoints. So I don't think this is an issue.

chamikaramj · 2023-09-21T22:21:08Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+    PTransform transformToUpgrade =
+        runnerAPIpipeline.getComponents().getTransformsMap().get(transformId);
+    if (transformToUpgrade == null) {
+      throw new Exception("Could not find a transform with the ID " + transformId);


Updated to throw IllegalArgumentException.

chamikaramj · 2023-09-21T22:23:44Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+      throw new RuntimeException(e);
+    }
+
+    Row configRow =


chamikaramj · 2023-09-21T22:32:22Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+        clientFactory.getExpansionServiceClient(transformServiceEndpoint).expand(request);
+
+    if (!Strings.isNullOrEmpty(response.getError())) {
+      throw new IOException(String.format("expansion service error: %s", response.getError()));


Updated to RuntimeException.

chamikaramj · 2023-09-21T23:11:43Z

...tion-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

+
+        ByteStringOutputStream schemaOutputStream = new ByteStringOutputStream();
+        try (ObjectOutputStream schemaObjOut = new ObjectOutputStream(schemaOutputStream)) {
+          schemaObjOut.writeObject(SchemaTranslation.schemaToProto(configRow.getSchema(), true));


chamikaramj · 2023-09-21T23:11:52Z

...tion-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java

+        // optional.
+      }
+      if (configRow != null) {
+        ByteStringOutputStream rowOutputStream = new ByteStringOutputStream();


chamikaramj · 2023-09-22T00:41:44Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+      }
+    }
+
+    String newTransformId = transformId + "_upgraded";


chamikaramj · 2023-09-22T00:41:50Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+                        transformBuilder.putAllInputs(updatedInputsMap);
+                      }
+
+                      // Fix sub-transforms


chamikaramj · 2023-09-22T00:41:56Z

...truction-java/src/main/java/org/apache/beam/runners/core/construction/TransformUpgrader.java

+    newComponentsBuilder.putAllTransforms(updatedExpandedTransformMap);
+    newComponentsBuilder.putTransforms(newTransformId, expandedTransform);
+
+    // We fix the root in case the overridden transform was one of the roots.


…ple times

chamikaramj · 2023-09-27T16:04:21Z

Friendly ping :)

robertwb

Thanks! This looks good to me.

chamikaramj · 2023-09-27T21:43:37Z

Thanks.

github-actions bot added java io runners dataflow gcp core labels Aug 29, 2023

chamikaramj changed the title ~~Upgrade Java transforms without upgrading the pipelines~~ Upgrade transforms without upgrading the pipelines Aug 29, 2023

robertwb reviewed Aug 31, 2023

View reviewed changes

chamikaramj force-pushed the cross_version_transform_upgrade branch from 1063fbf to c593512 Compare September 13, 2023 04:30

chamikaramj marked this pull request as ready for review September 13, 2023 04:31

chamikaramj commented Sep 13, 2023

View reviewed changes

robertwb reviewed Sep 15, 2023

View reviewed changes

chamikaramj force-pushed the cross_version_transform_upgrade branch 2 times, most recently from bc993a3 to d5fd742 Compare September 20, 2023 23:22

chamikaramj commented Sep 20, 2023

View reviewed changes

github-actions bot added spark flink samza labels Sep 21, 2023

chamikaramj added 5 commits September 21, 2023 13:48

Upgrade Java transforms without upgrading the pipelines

f3aa0ae

Addresses reviewer comments

3edbd62

Reduce visibility of the test-only constructor

7a4271f

Fix compile errors

e589f03

Fix spotless

6a85535

chamikaramj force-pushed the cross_version_transform_upgrade branch from d0fd5df to 6a85535 Compare September 21, 2023 21:00

robertwb reviewed Sep 21, 2023

View reviewed changes

Addressing reviewer comments

732bd32

chamikaramj commented Sep 22, 2023

View reviewed changes

Do not bundle Transform Service Launcher in the harness

3898c12

Fix harness build and a fix for when a runner invokes toProto() multi…

55b74b5

…ple times

robertwb approved these changes Sep 27, 2023

View reviewed changes

chamikaramj merged commit 1c50fd2 into apache:master Sep 27, 2023
26 of 29 checks passed

Upgrade transforms without upgrading the pipelines #28210

Upgrade transforms without upgrading the pipelines #28210

Conversation

chamikaramj commented Aug 29, 2023 • edited Loading

GitHub Actions Tests Status (on master branch)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj commented Sep 13, 2023

github-actions bot commented Sep 13, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robertwb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj commented Aug 29, 2023 •

edited

Loading