def test_combined_data_loader_with_max_size_cycle_and_ddp(monkeypatch, accelerator, use_distributed_sampler):
    """This test makes sure distributed sampler has been properly injected in dataloaders when using CombinedLoader
    with ddp and `max_size_cycle` mode."""
    trainer = Trainer(
        strategy="ddp", accelerator=accelerator, devices=2, use_distributed_sampler=use_distributed_sampler
    )
    model = BoringModel()

    combined_loader = CombinedLoader(
        {"a": DataLoader(RandomDataset(32, 8), batch_size=1), "b": DataLoader(RandomDataset(32, 8), batch_size=1)},
    )
    trainer.strategy.connect(model)
    trainer._data_connector.attach_data(model, train_dataloaders=combined_loader)
    trainer.fit_loop.setup_data()

    assert len(combined_loader) == 4 if use_distributed_sampler else 8

    for a_length in [6, 8, 10]:
        combined_loader = CombinedLoader(
            {
                "a": DataLoader(range(a_length), batch_size=1),
                "b": DataLoader(range(8), batch_size=1),
            },
            mode="max_size_cycle",
        )
        iter(combined_loader)
        length = max(a_length, 8)
        assert len(combined_loader) == length

        trainer._data_connector.attach_data(model, train_dataloaders=combined_loader)

        original_process_dataloader = trainer._data_connector._prepare_dataloader

        def non_shuffle_process_dataloader(dl, shuffle, mode):
            # avoid shuffling
            return original_process_dataloader(dl, False, mode)

        monkeypatch.setattr(trainer._data_connector, "_prepare_dataloader", non_shuffle_process_dataloader)
        trainer.fit_loop.setup_data()
        monkeypatch.undo()

        assert len(combined_loader) == length // 2 if use_distributed_sampler else length
        if use_distributed_sampler:
            last_batch = list(combined_loader)[-1][0]
            if a_length == 6:
                assert last_batch == {"a": torch.tensor([0]), "b": torch.tensor([6])}
            elif a_length == 8:
                assert last_batch == {"a": torch.tensor([6]), "b": torch.tensor([6])}
            elif a_length == 10:
                assert last_batch == {"a": torch.tensor([8]), "b": torch.tensor([0])}

    class InfiniteDataset(IterableDataset):
        def __iter__(self):
            while True:
                yield 1

    combined_loader = CombinedLoader(
        {
            "a": DataLoader(InfiniteDataset(), batch_size=1),
            "b": DataLoader(range(8), batch_size=1),
        },
        mode="max_size_cycle",
    )
    assert len(combined_loader.iterables["b"]) == 8

    trainer._data_connector.attach_data(model, train_dataloaders=combined_loader)
    trainer.fit_loop.setup_data()

    assert len(combined_loader.iterables["b"]) == 4 if use_distributed_sampler else 8
This test makes sure distributed sampler has been properly injected in dataloaders when using CombinedLoader
    with ddp and `max_size_cycle` mode.
tests/tests_pytorch/utilities/test_combined_loader.py