[Bug]: BigtableIO.readChangeStream() with RESUME_OR_NEW can start duplicate consumers inside one Dataflow job

### What happened?

## Environment

- Apache Beam Java SDK 2.75.0
- Google Cloud Dataflow / Portable Runner
- Exactly one active job and one worker
- Stable, explicitly configured `changeStreamName`

## Configuration

```java
BigtableIO.readChangeStream()
.withProjectId(projectId)
.withInstanceId(instanceId)
.withTableId(tableId)
.withAppProfileId(appProfileId)
.withMetadataTableTableId(metadataTableId)
.withChangeStreamName(changeStreamName)
.withExistingPipelineOptions(
BigtableIO.ExistingPipelineOptions.RESUME_OR_NEW);
```

## Observed behavior

1. The job starts normally and consumes every event once.
2. After approximately 30 minutes, the running job logs:

```text
Resuming from previous pipeline with low watermark of ...
```

3. Every subsequent event is then consumed exactly twice.
4. No second job is started and the worker count remains one.
5. The duplicate consumption is persistent, not just a short overlap during a checkpoint.
6. Omitting `withExistingPipelineOptions(...)`, thereby using the default `FAIL_IF_EXISTS`, prevents the problem.

## Expected behavior

Retries, checkpointing, or repeated initialization within the same active job must not cause it to resume itself or create another consumer for partitions already being consumed.

## Suspected cause

Dataflow can re-execute `InitializeDoFn`. After the current job has created its metadata, `RESUME_OR_NEW` appears to interpret that metadata as belonging to a previous pipeline. The observed `Resuming from previous pipeline...` log line indicates that this branch is executed while the original consumer is still active.

`ResumeFromPreviousPipelineAction` restores the existing partition UUIDs. The locking code in `MetadataTableDao.lockAndRecordPartition()` considers an existing lock with the same UUID successfully held:

```java
if (doHoldLock(
partitionRecord.getPartition(),
partitionRecord.getUuid())) {
return true;
}
```

The original and resumed readers can therefore both pass the lock check and open change-stream RPCs for the same partitions.

The workaround supports this explanation. With the default `FAIL_IF_EXISTS`, repeated initialization detects the existing metadata and returns without emitting another `InitialPipelineState`, preventing another downstream consumer chain from starting.

## Workaround

Omit `withExistingPipelineOptions(...)`. This prevents duplicate initialization, but it also prevents a legitimate later job from resuming existing metadata, so it is not a complete solution.

## Suggested fixes

1. Make initialization idempotent per job execution. Store a unique execution identifier in the metadata and claim initialization atomically. Re-execution with the same identifier should emit nothing, while a genuinely new job should still be allowed to resume.

2. Add a durable checkpoint or reshuffle boundary between `InitializeDoFn` and the unbounded `DetectNewPartitionsDoFn`. This should ensure that retries or backup executions of initialization cannot start multiple downstream consumer chains.

### Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

### Issue Components

- [ ] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [x] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner

Start by tracing InitializeDoFn and ResumeFromPreviousPipelineAction, then inspect MetadataTableDao.lockAndRecordPartition() and the RESUME_OR_NEW initialization path. Reproduce the repeated-initialization behavior with one active Dataflow job and verify that retries do not create duplicate consumers, while a genuinely new job can still resume existing metadata.

Written by the indexing model from the issue text.