@VisibleForTesting
  /*package*/ ExpansionApi.ExpansionResponse expand(ExpansionApi.ExpansionRequest request) {
    LOG.info(
        "Expanding '{}' with URN '{}'",
        request.getTransform().getUniqueName(),
        request.getTransform().getSpec().getUrn());
    LOG.debug("Full transform: {}", request.getTransform());
    Set<String> existingTransformIds = request.getComponents().getTransformsMap().keySet();
    Pipeline pipeline =
        createPipeline(PipelineOptionsTranslation.fromProto(request.getPipelineOptions()));
    boolean isUseDeprecatedRead =
        ExperimentalOptions.hasExperiment(pipelineOptions, "use_deprecated_read")
            || ExperimentalOptions.hasExperiment(
                pipelineOptions, "beam_fn_api_use_deprecated_read");
    if (!isUseDeprecatedRead) {
      ExperimentalOptions.addExperiment(
          pipeline.getOptions().as(ExperimentalOptions.class), "beam_fn_api");
      // TODO(https://github.com/apache/beam/issues/20530): Remove this when we address performance
      // issue.
      ExperimentalOptions.addExperiment(
          pipeline.getOptions().as(ExperimentalOptions.class), "use_sdf_read");
    } else {
      LOG.warn(
          "Using use_depreacted_read in portable runners is runner-dependent. The "
              + "ExpansionService will respect that, but if your runner does not have support for "
              + "native Read transform, your Pipeline will fail during Pipeline submission.");
    }

    RehydratedComponents rehydratedComponents =
        RehydratedComponents.forComponents(request.getComponents()).withPipeline(pipeline);

    Map<String, PCollection<?>> inputs =
        request.getTransform().getInputsMap().entrySet().stream()
            .collect(
                Collectors.toMap(
                    Map.Entry::getKey,
                    input -> {
                      try {
                        return rehydratedComponents.getPCollection(input.getValue());
                      } catch (IOException exn) {
                        throw new RuntimeException(exn);
                      }
                    }));

    String urn = request.getTransform().getSpec().getUrn();

    TransformProvider transformProvider = getRegisteredTransforms().get(urn);
    if (transformProvider == null) {
      if (getUrn(ExpansionMethods.Enum.JAVA_CLASS_LOOKUP).equals(urn)) {
        AllowList allowList =
            pipelineOptions.as(ExpansionServiceOptions.class).getJavaClassLookupAllowlist();
        assert allowList != null;
        transformProvider = new JavaClassLookupTransformProvider(allowList);
      } else if (getUrn(SCHEMA_TRANSFORM).equals(urn)) {
        try {
          String underlyingIdentifier =
              ExternalTransforms.SchemaTransformPayload.parseFrom(
                      request.getTransform().getSpec().getPayload())
                  .getIdentifier();
          transformProvider = getRegisteredTransforms().get(underlyingIdentifier);
        } catch (InvalidProtocolBufferException e) {
          throw new RuntimeException(e);
        }
        transformProvider =
            transformProvider != null
                ? transformProvider
                : ExpansionServiceSchemaTransformProvider.of();
      } else {
        throw new UnsupportedOperationException(
            "Unknown urn: " + request.getTransform().getSpec().getUrn());
      }
    }

    List<String> classpathResources =
        transformProvider.getDependencies(request.getTransform().getSpec(), pipeline.getOptions());
    pipeline.getOptions().as(PortablePipelineOptions.class).setFilesToStage(classpathResources);

    Map<String, PCollection<?>> outputs =
        transformProvider.apply(
            pipeline,
            request.getTransform().getUniqueName(),
            request.getTransform().getSpec(),
            inputs);

    // Needed to find which transform was new...
    SdkComponents sdkComponents =
        rehydratedComponents
            .getSdkComponents(request.getRequirementsList())
            .withNewIdPrefix(request.getNamespace());
    RunnerApi.Environment defaultEnvironment =
        Environments.createOrGetDefaultEnvironment(
            pipeline.getOptions().as(PortablePipelineOptions.class));
    if (pipelineOptions.as(ExpansionServiceOptions.class).getAlsoStartLoopbackWorker()) {
      PortablePipelineOptions externalOptions =
          PipelineOptionsFactory.create().as(PortablePipelineOptions.class);
      externalOptions.setDefaultEnvironmentType(Environments.ENVIRONMENT_EXTERNAL);
      externalOptions.setDefaultEnvironmentConfig(loopbackAddress);
      defaultEnvironment =
          Environments.createAnyOfEnvironment(
              defaultEnvironment, Environments.createOrGetDefaultEnvironment(externalOptions));
    }
    sdkComponents.registerEnvironment(defaultEnvironment);
    Map<String, String> outputMap =
        outputs.entrySet().stream()
            .collect(
                Collectors.toMap(
                    Map.Entry::getKey,
                    output -> {
                      try {
                        return sdkComponents.registerPCollection(output.getValue());
                      } catch (IOException exn) {
                        throw new RuntimeException(exn);
                      }
                    }));

    if (isUseDeprecatedRead) {
      SplittableParDo.convertReadBasedSplittableDoFnsToPrimitiveReadsIfNecessary(pipeline);
    }

    RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(pipeline, sdkComponents);
    String expandedTransformId =
        Iterables.getOnlyElement(
            pipelineProto.getRootTransformIdsList().stream()
                .filter(id -> !existingTransformIds.contains(id))
                .collect(Collectors.toList()));
    RunnerApi.Components components = pipelineProto.getComponents();
    RunnerApi.PTransform expandedTransform =
        components
            .getTransformsOrThrow(expandedTransformId)
            .toBuilder()
            .setUniqueName(expandedTransformId)
            .clearOutputs()
            .putAllOutputs(outputMap)
            .build();
    LOG.debug("Expanded to {}", expandedTransform);

    return ExpansionApi.ExpansionResponse.newBuilder()
        .setComponents(components.toBuilder().removeTransforms(expandedTransformId))
        .setTransform(expandedTransform)
        .addAllRequirements(pipelineProto.getRequirementsList())
        .build();
  }
@Test
  public void testConstructGenerateSequenceWithRegistration() {
    ExternalTransforms.ExternalConfigurationPayload payload =
        encodeRowIntoExternalConfigurationPayload(
            Row.withSchema(
                    Schema.of(
                        Field.of("start", FieldType.INT64),
                        Field.nullable("stop", FieldType.INT64)))
                .withFieldValue("start", 0L)
                .withFieldValue("stop", 1L)
                .build());

    Pipeline p = Pipeline.create();
    RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p);
    ExpansionApi.ExpansionRequest request =
        ExpansionApi.ExpansionRequest.newBuilder()
            .setComponents(pipelineProto.getComponents())
            .setTransform(
                RunnerApi.PTransform.newBuilder()
                    .setUniqueName(TEST_NAME)
                    .setSpec(
                        RunnerApi.FunctionSpec.newBuilder()
                            .setUrn(GenerateSequence.External.URN)
                            .setPayload(payload.toByteString())))
            .setNamespace(TEST_NAMESPACE)
            .build();
    ExpansionApi.ExpansionResponse response = expansionService.expand(request);
    RunnerApi.PTransform expandedTransform = response.getTransform();
    assertEquals(TEST_NAMESPACE + TEST_NAME, expandedTransform.getUniqueName());
    assertThat(expandedTransform.getInputsCount(), Matchers.is(0));
    assertThat(expandedTransform.getOutputsCount(), Matchers.is(1));
    assertThat(expandedTransform.getSubtransformsCount(), greaterThan(0));
  }