diff --git a/esrally/driver/driver.py b/esrally/driver/driver.py
index cfb50622..112471b4 100644
--- a/esrally/driver/driver.py
+++ b/esrally/driver/driver.py
@@ -1230,9 +1230,16 @@ def schedule_for(current_track, task, client_index):
                     "time period of [%s] seconds." % (task.schedule, task, str(warmup_time_period), str(task.time_period)))
         return time_period_based(sched, warmup_time_period, task.time_period, runner_for_op, params_for_op)
     else:
+        warmup_iterations = task.warmup_iterations if task.warmup_iterations else 0
+        if task.iterations:
+            iterations = task.iterations
+        elif params_for_op.size():
+            iterations = params_for_op.size() - warmup_iterations
+        else:
+            iterations = 1
         logger.info("Creating iteration-count based schedule with [%s] distribution for [%s] with [%d] warmup iterations and "
-                    "[%d] iterations." % (task.schedule, op, task.warmup_iterations, task.iterations))
-        return iteration_count_based(sched, task.warmup_iterations, task.iterations, runner_for_op, params_for_op)
+                    "[%d] iterations." % (task.schedule, op, warmup_iterations, iterations))
+        return iteration_count_based(sched, warmup_iterations, iterations, runner_for_op, params_for_op)
 
 
 def time_period_based(sched, warmup_time_period, time_period, runner, params):
diff --git a/esrally/track/loader.py b/esrally/track/loader.py
index 013a9875..c8e7b401 100644
--- a/esrally/track/loader.py
+++ b/esrally/track/loader.py
@@ -625,11 +625,11 @@ def post_process_for_test_mode(t):
             # we need iterate over leaf tasks and await iterating over possible intermediate 'parallel' elements
             for leaf_task in task:
                 # iteration-based schedules are divided among all clients and we should provide at least one iteration for each client.
-                if leaf_task.warmup_iterations > leaf_task.clients:
+                if leaf_task.warmup_iterations is not None and leaf_task.warmup_iterations > leaf_task.clients:
                     count = leaf_task.clients
                     logger.info("Resetting warmup iterations to %d for [%s]" % (count, str(leaf_task)))
                     leaf_task.warmup_iterations = count
-                if leaf_task.iterations > leaf_task.clients:
+                if leaf_task.iterations is not None and leaf_task.iterations > leaf_task.clients:
                     count = leaf_task.clients
                     logger.info("Resetting measurement iterations to %d for [%s]" % (count, str(leaf_task)))
                     leaf_task.iterations = count
@@ -1054,8 +1054,8 @@ class TrackSpecificationReader:
 
     def parse_parallel(self, ops_spec, ops, challenge_name):
         # use same default values as #parseTask() in case the 'parallel' element did not specify anything
-        default_warmup_iterations = self._r(ops_spec, "warmup-iterations", error_ctx="parallel", mandatory=False, default_value=0)
-        default_iterations = self._r(ops_spec, "iterations", error_ctx="parallel", mandatory=False, default_value=1)
+        default_warmup_iterations = self._r(ops_spec, "warmup-iterations", error_ctx="parallel", mandatory=False)
+        default_iterations = self._r(ops_spec, "iterations", error_ctx="parallel", mandatory=False)
         default_warmup_time_period = self._r(ops_spec, "warmup-time-period", error_ctx="parallel", mandatory=False)
         default_time_period = self._r(ops_spec, "time-period", error_ctx="parallel", mandatory=False)
         clients = self._r(ops_spec, "clients", error_ctx="parallel", mandatory=False)
@@ -1079,7 +1079,7 @@ class TrackSpecificationReader:
                             "this name exists." % (challenge_name, completed_by))
         return track.Parallel(tasks, clients)
 
-    def parse_task(self, task_spec, ops, challenge_name, default_warmup_iterations=0, default_iterations=1,
+    def parse_task(self, task_spec, ops, challenge_name, default_warmup_iterations=None, default_iterations=None,
                    default_warmup_time_period=None, default_time_period=None, completed_by_name=None):
 
         op_spec = task_spec["operation"]
@@ -1106,10 +1106,10 @@ class TrackSpecificationReader:
                           schedule=schedule,
                           # this is to provide scheduler-specific parameters for custom schedulers.
                           params=task_spec)
-        if task.warmup_iterations != default_warmup_iterations and task.time_period is not None:
+        if task.warmup_iterations is not None and task.time_period is not None:
             self._error("Operation '%s' in challenge '%s' defines '%d' warmup iterations and a time period of '%d' seconds. Please do not "
                         "mix time periods and iterations." % (op.name, challenge_name, task.warmup_iterations, task.time_period))
-        elif task.warmup_time_period is not None and task.iterations != default_iterations:
+        elif task.warmup_time_period is not None and task.iterations is not None:
             self._error("Operation '%s' in challenge '%s' defines a warmup time period of '%d' seconds and '%d' iterations. Please do not "
                         "mix time periods and iterations." % (op.name, challenge_name, task.warmup_time_period, task.iterations))
 
diff --git a/esrally/track/params.py b/esrally/track/params.py
index 65257f38..c8b662c2 100644
--- a/esrally/track/params.py
+++ b/esrally/track/params.py
@@ -113,8 +113,8 @@ class ParamSource:
         * It will either run an operation for a pre-determined number of times or
         * It can run until the parameter source is exhausted.
 
-        In the former case, return just 1. In the latter case, you should determine the number of times that `#params()` will be invoked.
-        With that number, Rally can show the progress made so far to the user.
+        In the former case, you should determine the number of times that `#params()` will be invoked. With that number, Rally can show
+        the progress made so far to the user. In the latter case, return ``None``.
 
         :return:  The "size" of this parameter source or ``None`` if should run eternally.
         """
diff --git a/esrally/track/track.py b/esrally/track/track.py
index 9aafbf8c..cfd1ba0a 100644
--- a/esrally/track/track.py
+++ b/esrally/track/track.py
@@ -588,7 +588,7 @@ class Parallel:
 
 
 class Task:
-    def __init__(self, name, operation, meta_data=None, warmup_iterations=0, iterations=1, warmup_time_period=None, time_period=None,
+    def __init__(self, name, operation, meta_data=None, warmup_iterations=None, iterations=None, warmup_time_period=None, time_period=None,
                  clients=1,
                  completes_parent=False, schedule="deterministic", params=None):
         self.name = name

Index-append operation only indexing bulk-size * clients documents
<!--

*******************************************************************************

Please ask questions in our forum at 

https://discuss.elastic.co/c/elasticsearch/rally

GitHub is reserved for bug reports and feature requests only and we will 
close your issue immediately if it is just a question.
 
*******************************************************************************

If you are in fact posting a bug report or a feature request, please include 
one and only one of the below blocks in your new issue.

-->

<!--
If you are filing a bug report, please remove the below feature
request block and provide responses for all of the below items.
-->

**Rally version** (get with `esrally --version`):
Latest from master, 425d8f6e8828f944607807d192aa7a98a45672f3

**Invoked command**:
```
./rally --track-path=/home/hinmanm/es/mytrack --target-hosts=127.0.0.1:9200 --pipeline=benchmark-only
```

**Configuration file (located in `~/.rally/rally.ini`))**:

```
[meta]
config.version = 12

[system]
env.name = local

[node]
root.dir = /home/hinmanm/.rally/benchmarks
src.root.dir = /home/hinmanm/es

[source]
remote.repo.url = https://github.com/elastic/elasticsearch.git
elasticsearch.src.subdir = elasticsearch

[build]
gradle.bin = /home/hinmanm/.sdkman/candidates/gradle/current/bin/gradle

[runtime]
java.home = /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.151-1.b12.fc26.x86_64

[benchmarks]
local.dataset.cache = ${node:root.dir}/data

[reporting]
datastore.type = elasticsearch
datastore.host = localhost
datastore.port = 9900
datastore.secure = False
datastore.user = 
datastore.password = 

[tracks]
default.url = https://github.com/elastic/rally-tracks

[teams]
default.url = https://github.com/elastic/rally-teams

[defaults]
preserve_benchmark_candidate = False

[distributions]
release.1.url = https://download.elasticsearch.org/elasticsearch/elasticsearch/elasticsearch-{{VERSION}}.tar.gz
release.2.url = https://download.elasticsearch.org/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/{{VERSION}}/elasticsearch-{{VERSION}}.tar.gz
release.url = https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-{{VERSION}}.tar.gz
release.cache = true
```

**JVM version**:
JDK 8

**OS version**:
Fedora 26

**Description of the problem including expected versus actual behavior**:

I have a track with an index-append operation defined inline in the challenge like so:

```json
      "schedule": [
        {
          "operation": {
            "name": "index-append",
            "operation-type": "bulk",
            "bulk-size": {{bulk_size | default(100)}}
          },
          "clients": 4
        },
```

The documents.json contains 1967 documents, however, only 400 are actually indexed.

**Steps to reproduce**:
 1. Using a track with many documents, add a challenge schedule with a low bulk-size and multiple clients
 2. Run the track
 3. Observe that only `bulk-size x clients` documents are indexed, in my case, `100 x 4 = 400` documents actually indexed.

I've noticed that this didn't affect me when the indexing was defined in a separate operation, it only started affecting me when I defined it inline in the challenge.

**Provide logs (if relevant)**:
The data is from a private repo, so I cannot provide it here.
diff --git a/tests/driver/driver_test.py b/tests/driver/driver_test.py
index fbfcc08a..981546e7 100644
--- a/tests/driver/driver_test.py
+++ b/tests/driver/driver_test.py
@@ -452,6 +452,47 @@ class SchedulerTests(ScheduleTestCase):
         ]
         self.assert_schedule(expected_schedule, schedule)
 
+    def test_schedule_param_source_determines_iterations_no_warmup(self):
+        # we neither define any time-period nor any iteration count on the task.
+        task = track.Task("bulk-index", track.Operation("bulk-index", track.OperationType.Bulk.name, params={"body": ["a"], "size": 3},
+                                                        param_source="driver-test-param-source"),
+                          clients=1, params={"target-throughput": 4, "clients": 4})
+
+        invocations = driver.schedule_for(self.test_track, task, 0)
+
+        self.assert_schedule([
+            (0.0, metrics.SampleType.Normal, 1 / 3, {"body": ["a"], "size": 3}),
+            (1.0, metrics.SampleType.Normal, 2 / 3, {"body": ["a"], "size": 3}),
+            (2.0, metrics.SampleType.Normal, 3 / 3, {"body": ["a"], "size": 3}),
+        ], list(invocations))
+
+    def test_schedule_param_source_determines_iterations_including_warmup(self):
+        task = track.Task("bulk-index", track.Operation("bulk-index", track.OperationType.Bulk.name, params={"body": ["a"], "size": 5},
+                                                        param_source="driver-test-param-source"),
+                          warmup_iterations=2, clients=1, params={"target-throughput": 4, "clients": 4})
+
+        invocations = driver.schedule_for(self.test_track, task, 0)
+
+        self.assert_schedule([
+            (0.0, metrics.SampleType.Warmup, 1 / 5, {"body": ["a"], "size": 5}),
+            (1.0, metrics.SampleType.Warmup, 2 / 5, {"body": ["a"], "size": 5}),
+            (2.0, metrics.SampleType.Normal, 3 / 5, {"body": ["a"], "size": 5}),
+            (3.0, metrics.SampleType.Normal, 4 / 5, {"body": ["a"], "size": 5}),
+            (4.0, metrics.SampleType.Normal, 5 / 5, {"body": ["a"], "size": 5}),
+        ], list(invocations))
+
+    def test_schedule_defaults_to_iteration_based(self):
+        # no time-period and no iterations specified on the task. Also, the parameter source does not define a size.
+        task = track.Task("bulk-index", track.Operation("bulk-index", track.OperationType.Bulk.name, params={"body": ["a"]},
+                                                        param_source="driver-test-param-source"),
+                          clients=1, params={"target-throughput": 4, "clients": 4})
+
+        invocations = driver.schedule_for(self.test_track, task, 0)
+
+        self.assert_schedule([
+            (0.0, metrics.SampleType.Normal, 1 / 1, {"body": ["a"]}),
+        ], list(invocations))
+
     def test_schedule_for_warmup_time_based(self):
         task = track.Task("time-based", track.Operation("time-based", track.OperationType.Bulk.name, params={"body": ["a"], "size": 11},
                                                         param_source="driver-test-param-source"),

pip install -e .
parse_log_pytest
pytest
pytest pytest-benchmark
apt-get update apt-get install -y gcc python3-pip python3-dev
3.6
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
attrs @ file:///opt/conda/conda-bld/attrs_1642510447205/work
certifi==2021.5.30
elasticsearch==6.0.0
-e git+https://github.com/elastic/rally.git@1ca746a1ef3cbf7c2441e9d6d2bb279795853991#egg=esrally
importlib-metadata @ file:///tmp/build/80754af9/importlib-metadata_1631916693255/work
iniconfig @ file:///home/linux1/recipes/ci/iniconfig_1610983019677/work
Jinja2==2.9.5
jsonschema==2.5.1
MarkupSafe==2.0.1
more-itertools @ file:///tmp/build/80754af9/more-itertools_1637733554872/work
packaging @ file:///tmp/build/80754af9/packaging_1637314298585/work
pluggy @ file:///tmp/build/80754af9/pluggy_1615976315926/work
psutil==5.4.0
py @ file:///opt/conda/conda-bld/py_1644396412707/work
py-cpuinfo==3.2.0
pyparsing @ file:///tmp/build/80754af9/pyparsing_1635766073266/work
pytest==6.2.4
pytest-benchmark==3.4.1
tabulate==0.8.1
thespian==3.9.2
toml @ file:///tmp/build/80754af9/toml_1616166611790/work
typing_extensions @ file:///opt/conda/conda-bld/typing_extensions_1647553014482/work
urllib3==1.22
zipp @ file:///tmp/build/80754af9/zipp_1633618647012/work

name: rally
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - attrs=21.4.0=pyhd3eb1b0_0
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - importlib-metadata=4.8.1=py36h06a4308_0
  - importlib_metadata=4.8.1=hd3eb1b0_0
  - iniconfig=1.1.1=pyhd3eb1b0_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - more-itertools=8.12.0=pyhd3eb1b0_0
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - packaging=21.3=pyhd3eb1b0_0
  - pip=21.2.2=py36h06a4308_0
  - pluggy=0.13.1=py36h06a4308_0
  - py=1.11.0=pyhd3eb1b0_0
  - pyparsing=3.0.4=pyhd3eb1b0_0
  - pytest=6.2.4=py36h06a4308_2
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - toml=0.10.2=pyhd3eb1b0_0
  - typing_extensions=4.1.1=pyh06a4308_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zipp=3.6.0=pyhd3eb1b0_0
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - elasticsearch==6.0.0
      - jinja2==2.9.5
      - jsonschema==2.5.1
      - markupsafe==2.0.1
      - psutil==5.4.0
      - py-cpuinfo==3.2.0
      - pytest-benchmark==3.4.1
      - tabulate==0.8.1
      - thespian==3.9.2
      - urllib3==1.22
prefix: /opt/conda/envs/rally

tests/driver/driver_test.py::SchedulerTests::test_schedule_param_source_determines_iterations_including_warmup tests/driver/driver_test.py::SchedulerTests::test_schedule_param_source_determines_iterations_no_warmup
tests/driver/driver_test.py::DriverTests::test_assign_drivers_round_robin tests/driver/driver_test.py::DriverTests::test_client_reaches_join_point_others_still_executing tests/driver/driver_test.py::DriverTests::test_client_reaches_join_point_which_completes_parent tests/driver/driver_test.py::DriverTests::test_start_benchmark_and_prepare_track tests/driver/driver_test.py::AllocatorTests::test_a_task_completes_the_parallel_structure tests/driver/driver_test.py::AllocatorTests::test_allocates_mixed_tasks tests/driver/driver_test.py::AllocatorTests::test_allocates_more_tasks_than_clients tests/driver/driver_test.py::AllocatorTests::test_allocates_one_task tests/driver/driver_test.py::AllocatorTests::test_allocates_two_parallel_tasks tests/driver/driver_test.py::AllocatorTests::test_allocates_two_serial_tasks tests/driver/driver_test.py::AllocatorTests::test_considers_number_of_clients_per_subtask tests/driver/driver_test.py::MetricsAggregationTests::test_different_sample_types tests/driver/driver_test.py::MetricsAggregationTests::test_single_metrics_aggregation tests/driver/driver_test.py::SchedulerTests::test_eternal_schedule_with_progress_indication tests/driver/driver_test.py::SchedulerTests::test_eternal_schedule_without_progress_indication tests/driver/driver_test.py::SchedulerTests::test_schedule_defaults_to_iteration_based tests/driver/driver_test.py::SchedulerTests::test_schedule_for_time_based tests/driver/driver_test.py::SchedulerTests::test_schedule_for_warmup_time_based tests/driver/driver_test.py::SchedulerTests::test_search_task_one_client tests/driver/driver_test.py::SchedulerTests::test_search_task_two_clients tests/driver/driver_test.py::ExecutorTests::test_cancel_execute_schedule tests/driver/driver_test.py::ExecutorTests::test_execute_schedule_aborts_on_error tests/driver/driver_test.py::ExecutorTests::test_execute_schedule_in_throughput_mode tests/driver/driver_test.py::ExecutorTests::test_execute_schedule_throughput_throttled tests/driver/driver_test.py::ExecutorTests::test_execute_single_dict tests/driver/driver_test.py::ExecutorTests::test_execute_single_no_return_value tests/driver/driver_test.py::ExecutorTests::test_execute_single_tuple tests/driver/driver_test.py::ExecutorTests::test_execute_single_with_connection_error tests/driver/driver_test.py::ExecutorTests::test_execute_single_with_http_400 tests/driver/driver_test.py::ExecutorTests::test_execute_single_with_key_error tests/driver/driver_test.py::ProfilerTests::test_profiler_is_a_transparent_wrapper
esrally/driver/driver.py esrally/track/params.py esrally/track/track.py esrally/track/loader.py
esrally/driver/driver.py esrally/track/params.py esrally/track/track.py esrally/track/loader.py