diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py
index f9fcb37..c9bbb97 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -835,12 +835,21 @@ class Fetcher(six.Iterator):
 
         return parsed_records
 
-    class PartitionRecords(six.Iterator):
+    class PartitionRecords(object):
         def __init__(self, fetch_offset, tp, messages):
             self.fetch_offset = fetch_offset
             self.topic_partition = tp
             self.messages = messages
-            self.message_idx = 0
+            # When fetching an offset that is in the middle of a
+            # compressed batch, we will get all messages in the batch.
+            # But we want to start 'take' at the fetch_offset
+            for i, msg in enumerate(messages):
+                if msg.offset == fetch_offset:
+                    self.message_idx = i
+                    break
+            else:
+                self.message_idx = 0
+                self.messages = None
 
         # For truthiness evaluation we need to define __len__ or __nonzero__
         def __len__(self):

Seek method returning incorrect messages on compressed topic when using max_poll_records
While using seek method of `kafka.consumer.group.seek' for a given partition, offset, we are seeing the inconsistent behavior for the messages returned with the subsequent poll method.
The issue is easily reproducible for the given topic (compacted).

Part of Workflow: 
```
from kafka.consumer.group import KafkaConsumer

topic_partition = TopicPartition(topic, 0)
consumer = KafkaConsumer(*consumer_config)
consumer.assign([topic_partition]) 
start_offset = 100 # Example value: highwatermark - 10
consumer.seek(partition=topic_partition, offset=start_offset)
messages = consumer.poll(timeout_ms=1000, max_records=1)[topic_partition]
message = messages[0]
print('Offset found:', message.offset, 'Expected offset:', start_offset)

Sample Output:
$ Offset found:80 Expected offset:100
```


Observation:
* If iterator interface is used instead of poll interface, the issue no longer exists. My guess is somewhere while polling for messages, the fetched offsets are not updated or fetched messages are not skipped. It looks like iterator method is not using fetched_records api that's why it works fine.

* At times it does give correct messages (especially when given offset is closer to highwatermark)

Please let me know if any other details are required.

diff --git a/test/test_fetcher.py b/test/test_fetcher.py
index 429071a..4547222 100644
--- a/test/test_fetcher.py
+++ b/test/test_fetcher.py
@@ -498,3 +498,43 @@ def test__parse_fetched_data__out_of_range(fetcher, topic, mocker):
     partition_record = fetcher._parse_fetched_data(completed_fetch)
     assert partition_record is None
     assert fetcher._subscriptions.assignment[tp].awaiting_reset is True
+
+
+def test_partition_records_offset():
+    """Test that compressed messagesets are handle correctly
+    when fetch offset is in the middle of the message list
+    """
+    batch_start = 120
+    batch_end = 130
+    fetch_offset = 123
+    tp = TopicPartition('foo', 0)
+    messages = [ConsumerRecord(tp.topic, tp.partition, i,
+                               None, None, 'key', 'value', 'checksum', 0, 0)
+                for i in range(batch_start, batch_end)]
+    records = Fetcher.PartitionRecords(fetch_offset, None, messages)
+    assert len(records) > 0
+    msgs = records.take(1)
+    assert msgs[0].offset == 123
+    assert records.fetch_offset == 124
+    msgs = records.take(2)
+    assert len(msgs) == 2
+    assert len(records) > 0
+    records.discard()
+    assert len(records) == 0
+
+
+def test_partition_records_empty():
+    records = Fetcher.PartitionRecords(0, None, [])
+    assert len(records) == 0
+
+
+def test_partition_records_no_fetch_offset():
+    batch_start = 0
+    batch_end = 100
+    fetch_offset = 123
+    tp = TopicPartition('foo', 0)
+    messages = [ConsumerRecord(tp.topic, tp.partition, i,
+                               None, None, 'key', 'value', 'checksum', 0, 0)
+                for i in range(batch_start, batch_end)]
+    records = Fetcher.PartitionRecords(fetch_offset, None, messages)
+    assert len(records) == 0

pip install -e .[dev]
parse_log_pytest
requirements.txt
pytest pytest-cov pytest-catchlog pytest-sugar pytest-mock mock python-snappy lz4 xxhash
apt-get update apt-get install -y libsnappy-dev
3.6
docs/requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
alabaster==0.7.13
attrs==22.2.0
Babel==2.11.0
certifi==2021.5.30
charset-normalizer==2.0.12
coverage==6.2
cramjam==2.5.0
docutils==0.18.1
idna==3.10
imagesize==1.4.1
importlib-metadata==4.8.3
iniconfig==1.1.1
Jinja2==3.0.3
-e git+https://github.com/dpkp/kafka-python.git@618c5051493693c1305aa9f08e8a0583d5fcf0e3#egg=kafka_python
lz4==3.1.10
MarkupSafe==2.0.1
mock==5.2.0
packaging==21.3
pluggy==1.0.0
pockets==0.9.1
py==1.11.0
Pygments==2.14.0
pyparsing==3.1.4
pytest==7.0.1
pytest-catchlog==1.2.2
pytest-cov==4.0.0
pytest-mock==3.6.1
pytest-sugar==0.9.6
python-snappy==0.7.3
pytz==2025.2
requests==2.27.1
six==1.17.0
snowballstemmer==2.2.0
Sphinx==5.3.0
sphinx-rtd-theme==2.0.0
sphinxcontrib-applehelp==1.0.2
sphinxcontrib-devhelp==1.0.2
sphinxcontrib-htmlhelp==2.0.0
sphinxcontrib-jquery==4.1
sphinxcontrib-jsmath==1.0.1
sphinxcontrib-napoleon==0.7
sphinxcontrib-qthelp==1.0.3
sphinxcontrib-serializinghtml==1.1.5
termcolor==1.1.0
tomli==1.2.3
typing_extensions==4.1.1
urllib3==1.26.20
xxhash==3.2.0
zipp==3.6.0

name: kafka-python
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
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_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
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - 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
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - alabaster==0.7.13
      - attrs==22.2.0
      - babel==2.11.0
      - charset-normalizer==2.0.12
      - coverage==6.2
      - cramjam==2.5.0
      - docutils==0.18.1
      - idna==3.10
      - imagesize==1.4.1
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - jinja2==3.0.3
      - lz4==3.1.10
      - markupsafe==2.0.1
      - mock==5.2.0
      - packaging==21.3
      - pluggy==1.0.0
      - pockets==0.9.1
      - py==1.11.0
      - pygments==2.14.0
      - pyparsing==3.1.4
      - pytest==7.0.1
      - pytest-catchlog==1.2.2
      - pytest-cov==4.0.0
      - pytest-mock==3.6.1
      - pytest-sugar==0.9.6
      - python-snappy==0.7.3
      - pytz==2025.2
      - requests==2.27.1
      - six==1.17.0
      - snowballstemmer==2.2.0
      - sphinx==5.3.0
      - sphinx-rtd-theme==2.0.0
      - sphinxcontrib-applehelp==1.0.2
      - sphinxcontrib-devhelp==1.0.2
      - sphinxcontrib-htmlhelp==2.0.0
      - sphinxcontrib-jquery==4.1
      - sphinxcontrib-jsmath==1.0.1
      - sphinxcontrib-napoleon==0.7
      - sphinxcontrib-qthelp==1.0.3
      - sphinxcontrib-serializinghtml==1.1.5
      - termcolor==1.1.0
      - tomli==1.2.3
      - typing-extensions==4.1.1
      - urllib3==1.26.20
      - xxhash==3.2.0
      - zipp==3.6.0
prefix: /opt/conda/envs/kafka-python

test/test_fetcher.py::test_partition_records_offset test/test_fetcher.py::test_partition_records_no_fetch_offset
test/test_fetcher.py::test_send_fetches test/test_fetcher.py::test_create_fetch_requests[api_version0-3] test/test_fetcher.py::test_create_fetch_requests[api_version1-2] test/test_fetcher.py::test_create_fetch_requests[api_version2-1] test/test_fetcher.py::test_create_fetch_requests[api_version3-0] test/test_fetcher.py::test_update_fetch_positions test/test_fetcher.py::test__reset_offset test/test_fetcher.py::test__send_offset_requests test/test_fetcher.py::test__send_offset_requests_multiple_nodes test/test_fetcher.py::test__handle_offset_response test/test_fetcher.py::test_fetched_records test/test_fetcher.py::test__handle_fetch_response[fetch_request0-fetch_response0-1] test/test_fetcher.py::test__handle_fetch_response[fetch_request1-fetch_response1-2] test/test_fetcher.py::test__handle_fetch_response[fetch_request2-fetch_response2-1] test/test_fetcher.py::test__handle_fetch_response[fetch_request3-fetch_response3-1] test/test_fetcher.py::test__handle_fetch_response[fetch_request4-fetch_response4-1] test/test_fetcher.py::test__handle_fetch_response[fetch_request5-fetch_response5-1] test/test_fetcher.py::test__unpack_message_set test/test_fetcher.py::test__message_generator test/test_fetcher.py::test__parse_fetched_data test/test_fetcher.py::test__parse_fetched_data__paused test/test_fetcher.py::test__parse_fetched_data__stale_offset test/test_fetcher.py::test__parse_fetched_data__not_leader test/test_fetcher.py::test__parse_fetched_data__unknown_tp test/test_fetcher.py::test__parse_fetched_data__out_of_range test/test_fetcher.py::test_partition_records_empty
kafka/consumer/fetcher.py
kafka/consumer/fetcher.py