Can we break fusion b/w ParDo using Windows + GroupBy or State & timely in batch pipeline of apache-beam?

### What happened?

**Context:**
I have N requests for which I need to place the fetch request (FetchData() ParDo) which in turn returns a result using which I can download the data (DownloadData() ParDo). Now these ParDo are getting fused due to which a single work place the fetch request & download the data then again place the request & download the data and so on.

So I want to parallelize these steps such that data starts to download as soon as I get the result from fetch step + fetch step to place another request while some data is getting downloaded in the next step.

Also I can't add `beam.Reshuffle() or beam.GroupBy()` before FetchData() ParDo because there is a limit to place fetch request to the client at a time (assume its 1 for now) but there is no limit on downloading from client in parallel.
So the issue I'm having is FetchData() outputs the element much faster then DownloadData() thats why I want to parallelize these two steps as download can be handled by some other worker while another request being place in background.

Runner: DirectRunner & DataflowRunner

**Attempt to break the fusion:**

request
| 'Fetch' >> beam.ParDo(FetchData())
| "GlobalWindow" >> beam.WindowInto(
window.GlobalWindows(),
trigger=trigger.Repeatedly(
trigger.AfterAny(
trigger.AfterCount(1)
)),
accumulation_mode=trigger.AccumulationMode.DISCARDING)
| 'GroupBy' >> beam.GroupBy()
| 'Download' >> beam.ParDo(DownloadData())

Actually I want to break the fusion w.r.t. FetchData() & DownloadData() ParDo, so I thought of this approach to have a GlobalWindows() & then use GroupBy() to group each window elements and send it further to DownloadData() ParDo while FetchData() ParDo works in parallel.

But what I'm observing here is that GroupBy() accumulates all the elements (waits for all the elements before its step to get processed first) before sending it further to DownloadData() ParDo.

Am I doing the right thing ? Anyway to make GroupBy() return early ? Or anyone have any other approach to achieve my goal ?

Update:

**Attempt-2 to break the fusion using states & timely:**

```
request
| 'Fetch' >> beam.ParDo())
| "SetRequestKey" >> beam.ParDo(SetRequestKeyFn())
| 'RequestBucket' >> beam.ParDo(RequestBucket())
| 'Download' >> beam.ParDo(DownloadData())

#Sets the request_id as the key
class SetRequestKeyFn(beam.DoFn):
def process(self, element):
return element[2]['href'], element

class RequestBucket(beam.DoFn):
"""Stateful ParDo for storing requests."""
REQUEST_STATE = userstate.BagStateSpec('requests', DillCoder())
EXPIRY_TIMER = userstate.TimerSpec('expiry_timer', userstate.TimeDomain.REAL_TIME)

def process(self,
element,
request_state=beam.DoFn.StateParam(REQUEST_STATE),
timer=beam.DoFn.TimerParam(EXPIRY_TIMER)):

logger.info(f"Adding new state {element[0]}.")
request_state.add(element)
# Set a timer to go off 0 seconds in the future.
timer.set(Timestamp.now() + Duration(seconds=0))

@userstate.on_timer(EXPIRY_TIMER)
def expiry_callback(self, request_state=beam.DoFn.StateParam(REQUEST_STATE)):
""""""
requests = list(request_state.read())
request_state.clear()
logger.info(f'Yielding for {requests!r}...')
yield requests[0]
```
Here also this `SetRequestKeyFn() ParDo` waits for all the elements before its step to get processed first before sending it further to `RequestBucket` ParDo.

**Attempt-3 to break the fusion using TimeStamp() & GroupByKey():**
Please refer to: https://github.com/apache/beam/issues/24665.

### Issue Priority

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

### Issue Components

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