@NonNull
  @Override
  public ConnectionFileName toPvfsFileName( @NonNull FileName providerFileName, @NonNull T details )
    throws KettleException {
    // Determine the part of provider file name following the connection "root".
    // Use the transformer to generate the connection root provider uri.
    // Both uris are assumed to be normalized.
    // Examples:
    // - connectionRootProviderUri: "hcp://domain.my:443/root/path/"           |  "s3://" |  "local://"
    // - providerUri:               "hcp://domain.my:443/root/path/rest/path"  |  "s3://rest/path"
    // Example: "pvfs://my-connection"

    String connectionRootProviderUri = getConnectionRootProviderUriPrefix( details );
    String providerUri = providerFileName.getURI();

    if ( !connectionFileNameUtils.isDescendantOrSelf( providerUri, connectionRootProviderUri ) ) {
      throw new IllegalArgumentException(
        String.format(
          "Provider file name '%s' is not a descendant of the connection root '%s'.",
          providerUri,
          connectionRootProviderUri ) );
    }

    String restUriPath = providerUri.substring( connectionRootProviderUri.length() );

    // Examples: "/rest/path" or "rest/path"

    return buildPvfsFileName( details, restUriPath, providerFileName.getType() );
  }
@Test
  public void testToPvfsFileNameHandlesConnectionsWithBucketsAndRootPath() throws Exception {
    // Example: S3 with root path

    when( details1.hasBuckets() ).thenReturn( true );
    mockDetailsWithRootPath( details1, "my/root/path" );

    String connectionRootProviderUriPrefix = "scheme1://my/root/path";
    String restPath = "/rest/path";

    FileName providerFileName = mockFileNameWithUri( FileName.class, connectionRootProviderUriPrefix + restPath );

    ConnectionFileName pvfsFileName = transformer.toPvfsFileName( providerFileName, details1 );

    assertEquals( "pvfs://connection-name1" + restPath, pvfsFileName.getURI() );

    // Should do connection root provider uri normalization.
    verify( kettleVFS, times( 1 ) ).resolveURI( connectionRootProviderUriPrefix );
  }