@Override
  public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset startMsgOffset, int timeoutMs) {
    long startOffset = ((LongMsgOffset) startMsgOffset).getOffset();
    if (LOGGER.isDebugEnabled()) {
      LOGGER.debug("Polling partition: {}, startOffset: {}, timeout: {}ms", _topicPartition, startOffset, timeoutMs);
    }
    if (_lastFetchedOffset < 0 || _lastFetchedOffset != startOffset - 1) {
      if (LOGGER.isDebugEnabled()) {
        LOGGER.debug("Seeking to offset: {}", startOffset);
      }
      _consumer.seek(_topicPartition, startOffset);
    }

    ConsumerRecords<String, Bytes> consumerRecords = _consumer.poll(Duration.ofMillis(timeoutMs));
    List<ConsumerRecord<String, Bytes>> records = consumerRecords.records(_topicPartition);
    List<BytesStreamMessage> filteredRecords = new ArrayList<>(records.size());
    long firstOffset = -1;
    long offsetOfNextBatch = startOffset;
    StreamMessageMetadata lastMessageMetadata = null;
    if (!records.isEmpty()) {
      firstOffset = records.get(0).offset();
      _lastFetchedOffset = records.get(records.size() - 1).offset();
      offsetOfNextBatch = _lastFetchedOffset + 1;
      for (ConsumerRecord<String, Bytes> record : records) {
        StreamMessageMetadata messageMetadata = extractMessageMetadata(record);
        Bytes message = record.value();
        if (message != null) {
          String key = record.key();
          byte[] keyBytes = key != null ? key.getBytes(StandardCharsets.UTF_8) : null;
          filteredRecords.add(new BytesStreamMessage(keyBytes, message.get(), messageMetadata));
        } else if (LOGGER.isDebugEnabled()) {
          LOGGER.debug("Tombstone message at offset: {}", record.offset());
        }
        lastMessageMetadata = messageMetadata;
      }
    }

    return new KafkaMessageBatch(filteredRecords, records.size(), offsetOfNextBatch, firstOffset, lastMessageMetadata,
        firstOffset > startOffset);
  }
@Test
  public void testOffsetsExpired()
      throws TimeoutException {
    Map<String, String> streamConfigMap = new HashMap<>();
    streamConfigMap.put("streamType", "kafka");
    streamConfigMap.put("stream.kafka.topic.name", TEST_TOPIC_3);
    streamConfigMap.put("stream.kafka.broker.list", _kafkaBrokerAddress);
    streamConfigMap.put("stream.kafka.consumer.type", "lowlevel");
    streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
    streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
    streamConfigMap.put("auto.offset.reset", "earliest");
    StreamConfig streamConfig = new StreamConfig("tableName_REALTIME", streamConfigMap);

    StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
    PartitionGroupConsumer consumer = streamConsumerFactory.createPartitionGroupConsumer("clientId",
        new PartitionGroupConsumptionStatus(0, 0, new LongMsgOffset(0),
            new LongMsgOffset(NUM_MSG_PRODUCED_PER_PARTITION), "CONSUMING"));

    // Start offset has expired. Automatically reset to earliest available and fetch whatever available
    MessageBatch messageBatch = consumer.fetchMessages(new LongMsgOffset(0), 10000);
    assertEquals(messageBatch.getMessageCount(), 500);
    assertEquals(messageBatch.getUnfilteredMessageCount(), 500);
    for (int i = 0; i < 500; i++) {
      assertEquals(new String((byte[]) messageBatch.getStreamMessage(i).getValue()), "sample_msg_" + (200 + i));
    }
    assertEquals(messageBatch.getOffsetOfNextBatch().toString(), "700");
  }