**Affected version:** 8.2 - 8.5.2
**Fixed in:** 8.5.3

### Problem Description

The problem applies to transforms that use more than 1 CCS cluster or 1 local and 1 CCS as source. A typical source could look like this (the important part: `remotes*`):

```
        "index" : [
          "remotes*:filebeat*"
        ],

```

If 2 indices share the same name the transform might fail with an error similar to this:

```
 Failed to retrieve checkpoint due to: Duplicate key logs-2022-11-14 (attempted merging values [... and ...]
```

Similar to the above the `_stats` call is broken for such a transform, it can fail with an error message starting with:

```
Failure during checkpoint retrieval
```

This can happen even if the transform is in state `STOPPED` or `FAILED`.

### Mitigation

Choose _one_ of the following mitigations:

 - split your transforms to use max 1 CCS/local indices

_or_

 - avoid upgrading to 8.2, until the issue got resolved (this is a regression of #80984)

_or_

 - avoid duplicate names for your indices even on remote cluster, e.g. name the index: `remote1-filebeat...`, `remote2-filebeat...`


### Solution

This is a regression of #80984 which introduced a new way of check-pointing. The remote needs to be prefixed to the index name.
x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/checkpoint/DefaultCheckpointProviderTests.java
GetIndexResponse getIndexResponse = new GetIndexResponse(new String[] { indexName }, null, null, null, null, null);
        doAnswer(withResponse(getIndexResponse)).when(client).execute(eq(GetIndexAction.INSTANCE), any(), any());
        IndicesStatsResponse indicesStatsResponse = mock(IndicesStatsResponse.class);
        doReturn(7).when(indicesStatsResponse).getFailedShards();
        doReturn(
            new DefaultShardOperationFailedException[] {
                new DefaultShardOperationFailedException(indexName, 3, new Exception("something's wrong")) }
        ).when(indicesStatsResponse).getShardFailures();
        doAnswer(withResponse(indicesStatsResponse)).when(client).execute(eq(IndicesStatsAction.INSTANCE), any(), any());
        DefaultCheckpointProvider provider = new DefaultCheckpointProvider(
            clock,
            client,
            remoteClusterResolver,
            transformConfigManager,
            transformAuditor,
            transformConfig
        );
        CountDownLatch latch = new CountDownLatch(1);
        provider.createNextCheckpoint(
            null,
            new LatchedActionListener<>(
                ActionListener.wrap(
                    response -> fail("This test case must fail"),
                    e -> assertThat(
                        e.getMessage(),
                        startsWith(
                            "Source has [7] failed shards, first shard failure: [some-index][3] failed, "
                                + "reason [java.lang.Exception: something's wrong"
                        )
                    )