text:
package kafka.server

import java.util.Optional

import kafka.api.Request
import kafka.cluster.{BrokerEndPoint, Partition}
import kafka.log.{Log, LogManager}
import kafka.server.AbstractFetcherThread.ResultWithPartitions
import kafka.server.QuotaFactory.UnboundedQuota
import kafka.utils.{DelayedItem, TestUtils}
import org.apache.kafka.common.errors.KafkaStorageException
import org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.OffsetForLeaderPartition
import org.apache.kafka.common.message.OffsetForLeaderEpochResponseData.EpochEndOffset
import org.apache.kafka.common.protocol.Errors
import org.apache.kafka.common.record.MemoryRecords
import org.apache.kafka.common.requests.FetchRequest
import org.apache.kafka.common.{IsolationLevel, TopicPartition}
import org.easymock.EasyMock._
import org.easymock.{Capture, CaptureType, EasyMock, IExpectationSetters}
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Test
import org.mockito.Mockito.{doNothing, when}
import org.mockito.{ArgumentCaptor, ArgumentMatchers, Mockito}

import scala.collection.{Map, Seq}
import scala.jdk.CollectionConverters._

class ReplicaAlterLogDirsThreadTest {

  private val t1p0 = new TopicPartition("topic1", 0)
  private val t1p1 = new TopicPartition("topic1", 1)
  private val failedPartitions = new FailedPartitions

  private def initialFetchState(fetchOffset: Long, leaderEpoch: Int = 1): InitialFetchState = {
    InitialFetchState(leader = new BrokerEndPoint(0, "localhost", 9092),
      initOffset = fetchOffset, currentLeaderEpoch = leaderEpoch)
  }

  @Test
  def shouldNotAddPartitionIfFutureLogIsNotDefined(): Unit = {
    val brokerId = 1
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId, "localhost:1234"))

    val replicaManager = Mockito.mock(classOf[ReplicaManager])
    val quotaManager = Mockito.mock(classOf[ReplicationQuotaManager])

    when(replicaManager.futureLogExists(t1p0)).thenReturn(false)

    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = quotaManager,
      brokerTopicStats = new BrokerTopicStats)

    val addedPartitions = thread.addPartitions(Map(t1p0 -> initialFetchState(0L)))
    assertEquals(Set.empty, addedPartitions)
    assertEquals(0, thread.partitionCount)
    assertEquals(None, thread.fetchState(t1p0))
  }

  @Test
  def shouldUpdateLeaderEpochAfterFencedEpochError(): Unit = {
    val brokerId = 1
    val partitionId = 0
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId, "localhost:1234"))

    val partition = Mockito.mock(classOf[Partition])
    val replicaManager = Mockito.mock(classOf[ReplicaManager])
    val quotaManager = Mockito.mock(classOf[ReplicationQuotaManager])
    val futureLog = Mockito.mock(classOf[Log])

    val leaderEpoch = 5
    val logEndOffset = 0

    when(partition.partitionId).thenReturn(partitionId)
    when(replicaManager.futureLocalLogOrException(t1p0)).thenReturn(futureLog)
    when(replicaManager.futureLogExists(t1p0)).thenReturn(true)
    when(replicaManager.onlinePartition(t1p0)).thenReturn(Some(partition))
    when(replicaManager.getPartitionOrException(t1p0)).thenReturn(partition)

    when(quotaManager.isQuotaExceeded).thenReturn(false)

    when(partition.lastOffsetForLeaderEpoch(Optional.empty(), leaderEpoch, fetchOnlyFromLeader = false))
      .thenReturn(new EpochEndOffset()
        .setPartition(partitionId)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpoch)
        .setEndOffset(logEndOffset))
    when(partition.futureLocalLogOrException).thenReturn(futureLog)
    doNothing().when(partition).truncateTo(offset = 0, isFuture = true)
    when(partition.maybeReplaceCurrentWithFutureReplica()).thenReturn(true)

    when(futureLog.logStartOffset).thenReturn(0L)
    when(futureLog.logEndOffset).thenReturn(0L)
    when(futureLog.latestEpoch).thenReturn(None)

    val fencedRequestData = new FetchRequest.PartitionData(0L, 0L,
      config.replicaFetchMaxBytes, Optional.of(leaderEpoch - 1))
    val fencedResponseData = FetchPartitionData(
      error = Errors.FENCED_LEADER_EPOCH,
      highWatermark = -1,
      logStartOffset = -1,
      records = MemoryRecords.EMPTY,
      divergingEpoch = None,
      lastStableOffset = None,
      abortedTransactions = None,
      preferredReadReplica = None,
      isReassignmentFetch = false)
    mockFetchFromCurrentLog(t1p0, fencedRequestData, config, replicaManager, fencedResponseData)

    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = quotaManager,
      brokerTopicStats = new BrokerTopicStats)

    // Initially we add the partition with an older epoch which results in an error
    thread.addPartitions(Map(t1p0 -> initialFetchState(fetchOffset = 0L, leaderEpoch - 1)))
    assertTrue(thread.fetchState(t1p0).isDefined)
    assertEquals(1, thread.partitionCount)

    thread.doWork()

    assertTrue(failedPartitions.contains(t1p0))
    assertEquals(None, thread.fetchState(t1p0))
    assertEquals(0, thread.partitionCount)

    // Next we update the epoch and assert that we can continue
    thread.addPartitions(Map(t1p0 -> initialFetchState(fetchOffset = 0L, leaderEpoch)))
    assertEquals(Some(leaderEpoch), thread.fetchState(t1p0).map(_.currentLeaderEpoch))
    assertEquals(1, thread.partitionCount)

    val requestData = new FetchRequest.PartitionData(0L, 0L,
      config.replicaFetchMaxBytes, Optional.of(leaderEpoch))
    val responseData = FetchPartitionData(
      error = Errors.NONE,
      highWatermark = 0L,
      logStartOffset = 0L,
      records = MemoryRecords.EMPTY,
      divergingEpoch = None,
      lastStableOffset = None,
      abortedTransactions = None,
      preferredReadReplica = None,
      isReassignmentFetch = false)
    mockFetchFromCurrentLog(t1p0, requestData, config, replicaManager, responseData)

    thread.doWork()

    assertFalse(failedPartitions.contains(t1p0))
    assertEquals(None, thread.fetchState(t1p0))
    assertEquals(0, thread.partitionCount)
  }

  @Test
  def shouldReplaceCurrentLogDirWhenCaughtUp(): Unit = {
    val brokerId = 1
    val partitionId = 0
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(brokerId, "localhost:1234"))

    val partition = Mockito.mock(classOf[Partition])
    val replicaManager = Mockito.mock(classOf[ReplicaManager])
    val quotaManager = Mockito.mock(classOf[ReplicationQuotaManager])
    val futureLog = Mockito.mock(classOf[Log])

    val leaderEpoch = 5
    val logEndOffset = 0

    when(partition.partitionId).thenReturn(partitionId)
    when(replicaManager.futureLocalLogOrException(t1p0)).thenReturn(futureLog)
    when(replicaManager.futureLogExists(t1p0)).thenReturn(true)
    when(replicaManager.onlinePartition(t1p0)).thenReturn(Some(partition))
    when(replicaManager.getPartitionOrException(t1p0)).thenReturn(partition)

    when(quotaManager.isQuotaExceeded).thenReturn(false)

    when(partition.lastOffsetForLeaderEpoch(Optional.empty(), leaderEpoch, fetchOnlyFromLeader = false))
      .thenReturn(new EpochEndOffset()
        .setPartition(partitionId)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpoch)
        .setEndOffset(logEndOffset))
    when(partition.futureLocalLogOrException).thenReturn(futureLog)
    doNothing().when(partition).truncateTo(offset = 0, isFuture = true)
    when(partition.maybeReplaceCurrentWithFutureReplica()).thenReturn(true)

    when(futureLog.logStartOffset).thenReturn(0L)
    when(futureLog.logEndOffset).thenReturn(0L)
    when(futureLog.latestEpoch).thenReturn(None)

    val requestData = new FetchRequest.PartitionData(0L, 0L,
      config.replicaFetchMaxBytes, Optional.of(leaderEpoch))
    val responseData = FetchPartitionData(
      error = Errors.NONE,
      highWatermark = 0L,
      logStartOffset = 0L,
      records = MemoryRecords.EMPTY,
      divergingEpoch = None,
      lastStableOffset = None,
      abortedTransactions = None,
      preferredReadReplica = None,
      isReassignmentFetch = false)
    mockFetchFromCurrentLog(t1p0, requestData, config, replicaManager, responseData)

    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = quotaManager,
      brokerTopicStats = new BrokerTopicStats)

    thread.addPartitions(Map(t1p0 -> initialFetchState(fetchOffset = 0L, leaderEpoch)))
    assertTrue(thread.fetchState(t1p0).isDefined)
    assertEquals(1, thread.partitionCount)

    thread.doWork()

    assertEquals(None, thread.fetchState(t1p0))
    assertEquals(0, thread.partitionCount)
  }

  private def mockFetchFromCurrentLog(topicPartition: TopicPartition,
                                      requestData: FetchRequest.PartitionData,
                                      config: KafkaConfig,
                                      replicaManager: ReplicaManager,
                                      responseData: FetchPartitionData): Unit = {
    val callbackCaptor: ArgumentCaptor[Seq[(TopicPartition, FetchPartitionData)] => Unit] =
      ArgumentCaptor.forClass(classOf[Seq[(TopicPartition, FetchPartitionData)] => Unit])
    when(replicaManager.fetchMessages(
      timeout = ArgumentMatchers.eq(0L),
      replicaId = ArgumentMatchers.eq(Request.FutureLocalReplicaId),
      fetchMinBytes = ArgumentMatchers.eq(0),
      fetchMaxBytes = ArgumentMatchers.eq(config.replicaFetchResponseMaxBytes),
      hardMaxBytesLimit = ArgumentMatchers.eq(false),
      fetchInfos = ArgumentMatchers.eq(Seq(topicPartition -> requestData)),
      quota = ArgumentMatchers.eq(UnboundedQuota),
      responseCallback = callbackCaptor.capture(),
      isolationLevel = ArgumentMatchers.eq(IsolationLevel.READ_UNCOMMITTED),
      clientMetadata = ArgumentMatchers.eq(None)
    )).thenAnswer(_ => {
      callbackCaptor.getValue.apply(Seq((topicPartition, responseData)))
    })
  }

  @Test
  def issuesEpochRequestFromLocalReplica(): Unit = {
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234"))

    //Setup all dependencies

    val partitionT1p0: Partition = createMock(classOf[Partition])
    val partitionT1p1: Partition = createMock(classOf[Partition])
    val replicaManager: ReplicaManager = createMock(classOf[ReplicaManager])

    val partitionT1p0Id = 0
    val partitionT1p1Id = 1
    val leaderEpochT1p0 = 2
    val leaderEpochT1p1 = 5
    val leoT1p0 = 13
    val leoT1p1 = 232

    //Stubs
    expect(partitionT1p0.partitionId).andStubReturn(partitionT1p0Id)
    expect(partitionT1p0.partitionId).andStubReturn(partitionT1p1Id)

    expect(replicaManager.getPartitionOrException(t1p0))
      .andStubReturn(partitionT1p0)
    expect(partitionT1p0.lastOffsetForLeaderEpoch(Optional.empty(), leaderEpochT1p0, fetchOnlyFromLeader = false))
      .andReturn(new EpochEndOffset()
        .setPartition(partitionT1p0Id)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpochT1p0)
        .setEndOffset(leoT1p0))
      .anyTimes()

    expect(replicaManager.getPartitionOrException(t1p1))
      .andStubReturn(partitionT1p1)
    expect(partitionT1p1.lastOffsetForLeaderEpoch(Optional.empty(), leaderEpochT1p1, fetchOnlyFromLeader = false))
      .andReturn(new EpochEndOffset()
        .setPartition(partitionT1p1Id)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpochT1p1)
        .setEndOffset(leoT1p1))
      .anyTimes()

    replay(partitionT1p0, partitionT1p1, replicaManager)

    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread-test1",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = null,
      brokerTopicStats = null)

    val result = thread.fetchEpochEndOffsets(Map(
      t1p0 -> new OffsetForLeaderPartition()
        .setPartition(t1p0.partition)
        .setLeaderEpoch(leaderEpochT1p0),
      t1p1 -> new OffsetForLeaderPartition()
        .setPartition(t1p1.partition)
        .setLeaderEpoch(leaderEpochT1p1)))

    val expected = Map(
      t1p0 -> new EpochEndOffset()
        .setPartition(t1p0.partition)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpochT1p0)
        .setEndOffset(leoT1p0),
      t1p1 -> new EpochEndOffset()
        .setPartition(t1p1.partition)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpochT1p1)
        .setEndOffset(leoT1p1)
    )

    assertEquals(expected, result, "results from leader epoch request should have offset from local replica")
  }

  @Test
  def fetchEpochsFromLeaderShouldHandleExceptionFromGetLocalReplica(): Unit = {
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234"))

    //Setup all dependencies
    val partitionT1p0: Partition = createMock(classOf[Partition])
    val replicaManager: ReplicaManager = createMock(classOf[ReplicaManager])

    val partitionId = 0
    val leaderEpoch = 2
    val leo = 13

    //Stubs
    expect(partitionT1p0.partitionId).andStubReturn(partitionId)

    expect(replicaManager.getPartitionOrException(t1p0))
      .andStubReturn(partitionT1p0)
    expect(partitionT1p0.lastOffsetForLeaderEpoch(Optional.empty(), leaderEpoch, fetchOnlyFromLeader = false))
      .andReturn(new EpochEndOffset()
        .setPartition(partitionId)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpoch)
        .setEndOffset(leo))
      .anyTimes()

    expect(replicaManager.getPartitionOrException(t1p1))
      .andThrow(new KafkaStorageException).once()

    replay(partitionT1p0, replicaManager)

    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread-test1",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = null,
      brokerTopicStats = null)

    val result = thread.fetchEpochEndOffsets(Map(
      t1p0 -> new OffsetForLeaderPartition()
        .setPartition(t1p0.partition)
        .setLeaderEpoch(leaderEpoch),
      t1p1 -> new OffsetForLeaderPartition()
        .setPartition(t1p1.partition)
        .setLeaderEpoch(leaderEpoch)))

    val expected = Map(
      t1p0 -> new EpochEndOffset()
        .setPartition(t1p0.partition)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpoch)
        .setEndOffset(leo),
      t1p1 -> new EpochEndOffset()
        .setPartition(t1p1.partition)
        .setErrorCode(Errors.KAFKA_STORAGE_ERROR.code)
    )

    assertEquals(expected, result)
  }

  @Test
  def shouldTruncateToReplicaOffset(): Unit = {

    //Create a capture to track what partitions/offsets are truncated
    val truncateCaptureT1p0: Capture[Long] = newCapture(CaptureType.ALL)
    val truncateCaptureT1p1: Capture[Long] = newCapture(CaptureType.ALL)

    // Setup all the dependencies
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234"))
    val quotaManager: ReplicationQuotaManager = createNiceMock(classOf[ReplicationQuotaManager])
    val logManager: LogManager = createMock(classOf[LogManager])
    val logT1p0: Log = createNiceMock(classOf[Log])
    val logT1p1: Log = createNiceMock(classOf[Log])
    // one future replica mock because our mocking methods return same values for both future replicas
    val futureLogT1p0: Log = createNiceMock(classOf[Log])
    val futureLogT1p1: Log = createNiceMock(classOf[Log])
    val partitionT1p0: Partition = createMock(classOf[Partition])
    val partitionT1p1: Partition = createMock(classOf[Partition])
    val replicaManager: ReplicaManager = createMock(classOf[ReplicaManager])
    val responseCallback: Capture[Seq[(TopicPartition, FetchPartitionData)] => Unit]  = EasyMock.newCapture()

    val partitionT1p0Id = 0
    val partitionT1p1Id = 1
    val leaderEpoch = 2
    val futureReplicaLEO = 191
    val replicaT1p0LEO = 190
    val replicaT1p1LEO = 192

    //Stubs
    expect(partitionT1p0.partitionId).andStubReturn(partitionT1p0Id)
    expect(partitionT1p1.partitionId).andStubReturn(partitionT1p1Id)

    expect(replicaManager.getPartitionOrException(t1p0))
      .andStubReturn(partitionT1p0)
    expect(replicaManager.getPartitionOrException(t1p1))
      .andStubReturn(partitionT1p1)
    expect(replicaManager.futureLocalLogOrException(t1p0)).andStubReturn(futureLogT1p0)
    expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true)
    expect(replicaManager.futureLocalLogOrException(t1p1)).andStubReturn(futureLogT1p1)
    expect(replicaManager.futureLogExists(t1p1)).andStubReturn(true)
    expect(partitionT1p0.truncateTo(capture(truncateCaptureT1p0), anyBoolean())).anyTimes()
    expect(partitionT1p1.truncateTo(capture(truncateCaptureT1p1), anyBoolean())).anyTimes()

    expect(futureLogT1p0.logEndOffset).andReturn(futureReplicaLEO).anyTimes()
    expect(futureLogT1p1.logEndOffset).andReturn(futureReplicaLEO).anyTimes()

    expect(futureLogT1p0.latestEpoch).andReturn(Some(leaderEpoch)).anyTimes()
    expect(futureLogT1p0.endOffsetForEpoch(leaderEpoch)).andReturn(
      Some(OffsetAndEpoch(futureReplicaLEO, leaderEpoch))).anyTimes()
    expect(partitionT1p0.lastOffsetForLeaderEpoch(Optional.of(1), leaderEpoch, fetchOnlyFromLeader = false))
      .andReturn(new EpochEndOffset()
        .setPartition(partitionT1p0Id)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpoch)
        .setEndOffset(replicaT1p0LEO))
      .anyTimes()

    expect(futureLogT1p1.latestEpoch).andReturn(Some(leaderEpoch)).anyTimes()
    expect(futureLogT1p1.endOffsetForEpoch(leaderEpoch)).andReturn(
      Some(OffsetAndEpoch(futureReplicaLEO, leaderEpoch))).anyTimes()
    expect(partitionT1p1.lastOffsetForLeaderEpoch(Optional.of(1), leaderEpoch, fetchOnlyFromLeader = false))
      .andReturn(new EpochEndOffset()
        .setPartition(partitionT1p1Id)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpoch)
        .setEndOffset(replicaT1p1LEO))
      .anyTimes()

    expect(replicaManager.logManager).andReturn(logManager).anyTimes()
    stubWithFetchMessages(logT1p0, logT1p1, futureLogT1p0, partitionT1p0, replicaManager, responseCallback)

    replay(replicaManager, logManager, quotaManager, partitionT1p0, partitionT1p1, logT1p0, logT1p1, futureLogT1p0, futureLogT1p1)

    //Create the thread
    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread-test1",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = quotaManager,
      brokerTopicStats = null)
    thread.addPartitions(Map(t1p0 -> initialFetchState(0L), t1p1 -> initialFetchState(0L)))

    //Run it
    thread.doWork()

    //We should have truncated to the offsets in the response
    assertEquals(replicaT1p0LEO, truncateCaptureT1p0.getValue)
    assertEquals(futureReplicaLEO, truncateCaptureT1p1.getValue)
  }

  @Test
  def shouldTruncateToEndOffsetOfLargestCommonEpoch(): Unit = {

    //Create a capture to track what partitions/offsets are truncated
    val truncateToCapture: Capture[Long] = newCapture(CaptureType.ALL)

    // Setup all the dependencies
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234"))
    val quotaManager: ReplicationQuotaManager = createNiceMock(classOf[ReplicationQuotaManager])
    val logManager: LogManager = createMock(classOf[LogManager])
    val log: Log = createNiceMock(classOf[Log])
    // one future replica mock because our mocking methods return same values for both future replicas
    val futureLog: Log = createNiceMock(classOf[Log])
    val partition: Partition = createMock(classOf[Partition])
    val replicaManager: ReplicaManager = createMock(classOf[ReplicaManager])
    val responseCallback: Capture[Seq[(TopicPartition, FetchPartitionData)] => Unit]  = EasyMock.newCapture()

    val partitionId = 0
    val leaderEpoch = 5
    val futureReplicaLEO = 195
    val replicaLEO = 200
    val replicaEpochEndOffset = 190
    val futureReplicaEpochEndOffset = 191

    //Stubs
    expect(partition.partitionId).andStubReturn(partitionId)

    expect(replicaManager.getPartitionOrException(t1p0))
      .andStubReturn(partition)
    expect(replicaManager.futureLocalLogOrException(t1p0)).andStubReturn(futureLog)
    expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true)

    expect(partition.truncateTo(capture(truncateToCapture), EasyMock.eq(true))).anyTimes()
    expect(futureLog.logEndOffset).andReturn(futureReplicaLEO).anyTimes()
    expect(futureLog.latestEpoch).andReturn(Some(leaderEpoch)).once()
    expect(futureLog.latestEpoch).andReturn(Some(leaderEpoch - 2)).times(3)

    // leader replica truncated and fetched new offsets with new leader epoch
    expect(partition.lastOffsetForLeaderEpoch(Optional.of(1), leaderEpoch, fetchOnlyFromLeader = false))
      .andReturn(new EpochEndOffset()
        .setPartition(partitionId)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpoch - 1)
        .setEndOffset(replicaLEO))
      .anyTimes()
    // but future replica does not know about this leader epoch, so returns a smaller leader epoch
    expect(futureLog.endOffsetForEpoch(leaderEpoch - 1)).andReturn(
      Some(OffsetAndEpoch(futureReplicaLEO, leaderEpoch - 2))).anyTimes()
    // finally, the leader replica knows about the leader epoch and returns end offset
    expect(partition.lastOffsetForLeaderEpoch(Optional.of(1), leaderEpoch - 2, fetchOnlyFromLeader = false))
      .andReturn(new EpochEndOffset()
        .setPartition(partitionId)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(leaderEpoch - 2)
        .setEndOffset(replicaEpochEndOffset))
      .anyTimes()
    expect(futureLog.endOffsetForEpoch(leaderEpoch - 2)).andReturn(
      Some(OffsetAndEpoch(futureReplicaEpochEndOffset, leaderEpoch - 2))).anyTimes()

    expect(replicaManager.logManager).andReturn(logManager).anyTimes()
    stubWithFetchMessages(log, null, futureLog, partition, replicaManager, responseCallback)

    replay(replicaManager, logManager, quotaManager, partition, log, futureLog)

    //Create the thread
    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread-test1",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = quotaManager,
      brokerTopicStats = null)
    thread.addPartitions(Map(t1p0 -> initialFetchState(0L)))

    // First run will result in another offset for leader epoch request
    thread.doWork()
    // Second run should actually truncate
    thread.doWork()

    //We should have truncated to the offsets in the response
    assertTrue(truncateToCapture.getValues.asScala.contains(replicaEpochEndOffset),
               "Expected offset " + replicaEpochEndOffset + " in captured truncation offsets " + truncateToCapture.getValues)
  }

  @Test
  def shouldTruncateToInitialFetchOffsetIfReplicaReturnsUndefinedOffset(): Unit = {

    //Create a capture to track what partitions/offsets are truncated
    val truncated: Capture[Long] = newCapture(CaptureType.ALL)

    // Setup all the dependencies
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234"))
    val quotaManager: ReplicationQuotaManager = createNiceMock(classOf[ReplicationQuotaManager])
    val logManager: LogManager = createMock(classOf[LogManager])
    val log: Log = createNiceMock(classOf[Log])
    val futureLog: Log = createNiceMock(classOf[Log])
    val partition: Partition = createMock(classOf[Partition])
    val replicaManager: ReplicaManager = createMock(classOf[ReplicaManager])
    val responseCallback: Capture[Seq[(TopicPartition, FetchPartitionData)] => Unit]  = EasyMock.newCapture()

    val initialFetchOffset = 100

    //Stubs
    expect(replicaManager.getPartitionOrException(t1p0))
      .andStubReturn(partition)
    expect(partition.truncateTo(capture(truncated), isFuture = EasyMock.eq(true))).anyTimes()
    expect(replicaManager.futureLocalLogOrException(t1p0)).andStubReturn(futureLog)
    expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true)

    expect(replicaManager.logManager).andReturn(logManager).anyTimes()

    // pretend this is a completely new future replica, with no leader epochs recorded
    expect(futureLog.latestEpoch).andReturn(None).anyTimes()

    stubWithFetchMessages(log, null, futureLog, partition, replicaManager, responseCallback)
    replay(replicaManager, logManager, quotaManager, partition, log, futureLog)

    //Create the thread
    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread-test1",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = quotaManager,
      brokerTopicStats = null)
    thread.addPartitions(Map(t1p0 -> initialFetchState(initialFetchOffset)))

    //Run it
    thread.doWork()

    //We should have truncated to initial fetch offset
    assertEquals(initialFetchOffset,
                 truncated.getValue, "Expected future replica to truncate to initial fetch offset if replica returns UNDEFINED_EPOCH_OFFSET")
  }

  @Test
  def shouldPollIndefinitelyIfReplicaNotAvailable(): Unit = {

    //Create a capture to track what partitions/offsets are truncated
    val truncated: Capture[Long] = newCapture(CaptureType.ALL)

    // Setup all the dependencies
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234"))
    val quotaManager: ReplicationQuotaManager = createNiceMock(classOf[ReplicationQuotaManager])
    val logManager: LogManager = createMock(classOf[LogManager])
    val log: Log = createNiceMock(classOf[Log])
    val futureLog: Log = createNiceMock(classOf[Log])
    val partition: Partition = createMock(classOf[Partition])
    val replicaManager: ReplicaManager = createMock(classOf[ReplicaManager])
    val responseCallback: Capture[Seq[(TopicPartition, FetchPartitionData)] => Unit]  = EasyMock.newCapture()

    val partitionId = 0
    val futureReplicaLeaderEpoch = 1
    val futureReplicaLEO = 290
    val replicaLEO = 300

    //Stubs
    expect(partition.partitionId).andStubReturn(partitionId)

    expect(replicaManager.getPartitionOrException(t1p0))
      .andStubReturn(partition)
    expect(partition.truncateTo(capture(truncated), isFuture = EasyMock.eq(true))).once()

    expect(replicaManager.futureLocalLogOrException(t1p0)).andStubReturn(futureLog)
    expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true)
    expect(futureLog.logEndOffset).andReturn(futureReplicaLEO).anyTimes()
    expect(futureLog.latestEpoch).andStubReturn(Some(futureReplicaLeaderEpoch))
    expect(futureLog.endOffsetForEpoch(futureReplicaLeaderEpoch)).andReturn(
      Some(OffsetAndEpoch(futureReplicaLEO, futureReplicaLeaderEpoch)))
    expect(replicaManager.localLog(t1p0)).andReturn(Some(log)).anyTimes()

    // this will cause fetchEpochsFromLeader return an error with undefined offset
    expect(partition.lastOffsetForLeaderEpoch(Optional.of(1), futureReplicaLeaderEpoch, fetchOnlyFromLeader = false))
      .andReturn(new EpochEndOffset()
        .setPartition(partitionId)
        .setErrorCode(Errors.REPLICA_NOT_AVAILABLE.code))
      .times(3)
      .andReturn(new EpochEndOffset()
        .setPartition(partitionId)
        .setErrorCode(Errors.NONE.code)
        .setLeaderEpoch(futureReplicaLeaderEpoch)
        .setEndOffset(replicaLEO))

    expect(replicaManager.logManager).andReturn(logManager).anyTimes()
    expect(replicaManager.fetchMessages(
      EasyMock.anyLong(),
      EasyMock.anyInt(),
      EasyMock.anyInt(),
      EasyMock.anyInt(),
      EasyMock.anyObject(),
      EasyMock.anyObject(),
      EasyMock.anyObject(),
      EasyMock.capture(responseCallback),
      EasyMock.anyObject(),
      EasyMock.anyObject())
    ).andAnswer(() => responseCallback.getValue.apply(Seq.empty[(TopicPartition, FetchPartitionData)])).anyTimes()

    replay(replicaManager, logManager, quotaManager, partition, log, futureLog)

    //Create the thread
    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread-test1",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = quotaManager,
      brokerTopicStats = null)
    thread.addPartitions(Map(t1p0 -> initialFetchState(0L)))

    // Run thread 3 times (exactly number of times we mock exception for getReplicaOrException)
    (0 to 2).foreach { _ =>
      thread.doWork()
    }

    // Nothing happened since the replica was not available
    assertEquals(0, truncated.getValues.size())

    // Next time we loop, getReplicaOrException will return replica
    thread.doWork()

    // Now the final call should have actually done a truncation (to offset futureReplicaLEO)
    assertEquals(futureReplicaLEO, truncated.getValue)
  }

  @Test
  def shouldFetchLeaderEpochOnFirstFetchOnly(): Unit = {

    //Setup all dependencies
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234"))
    val quotaManager: ReplicationQuotaManager = createNiceMock(classOf[ReplicationQuotaManager])
    val logManager: LogManager = createMock(classOf[LogManager])
    val log: Log = createNiceMock(classOf[Log])
    val futureLog: Log = createNiceMock(classOf[Log])
    val partition: Partition = createMock(classOf[Partition])
    val replicaManager: ReplicaManager = createMock(classOf[ReplicaManager])
    val responseCallback: Capture[Seq[(TopicPartition, FetchPartitionData)] => Unit]  = EasyMock.newCapture()

    val partitionId = 0
    val leaderEpoch = 5
    val futureReplicaLEO = 190
    val replicaLEO = 213

    expect(partition.partitionId).andStubReturn(partitionId)

    expect(replicaManager.getPartitionOrException(t1p0))
        .andStubReturn(partition)
    expect(partition.lastOffsetForLeaderEpoch(Optional.of(1), leaderEpoch, fetchOnlyFromLeader = false))
        .andReturn(new EpochEndOffset()
          .setPartition(partitionId)
          .setErrorCode(Errors.NONE.code)
          .setLeaderEpoch(leaderEpoch)
          .setEndOffset(replicaLEO))
    expect(partition.truncateTo(futureReplicaLEO, isFuture = true)).once()

    expect(replicaManager.futureLocalLogOrException(t1p0)).andStubReturn(futureLog)
    expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true)
    expect(futureLog.latestEpoch).andStubReturn(Some(leaderEpoch))
    expect(futureLog.logEndOffset).andStubReturn(futureReplicaLEO)
    expect(futureLog.endOffsetForEpoch(leaderEpoch)).andReturn(
      Some(OffsetAndEpoch(futureReplicaLEO, leaderEpoch)))
    expect(replicaManager.logManager).andReturn(logManager).anyTimes()
    stubWithFetchMessages(log, null, futureLog, partition, replicaManager, responseCallback)

    replay(replicaManager, logManager, quotaManager, partition, log, futureLog)

    //Create the fetcher thread
    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread-test1",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = quotaManager,
      brokerTopicStats = null)
    thread.addPartitions(Map(t1p0 -> initialFetchState(0L)))

    // loop few times
    (0 to 3).foreach { _ =>
      thread.doWork()
    }

    //Assert that truncate to is called exactly once (despite more loops)
    verify(partition)
  }

  @Test
  def shouldFetchOneReplicaAtATime(): Unit = {

    //Setup all dependencies
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234"))
    val quotaManager: ReplicationQuotaManager = createNiceMock(classOf[ReplicationQuotaManager])
    val logManager: LogManager = createMock(classOf[LogManager])
    val log: Log = createNiceMock(classOf[Log])
    val futureLog: Log = createNiceMock(classOf[Log])
    val partition: Partition = createMock(classOf[Partition])
    val replicaManager: ReplicaManager = createMock(classOf[ReplicaManager])

    //Stubs
    expect(replicaManager.logManager).andReturn(logManager).anyTimes()
    stub(log, null, futureLog, partition, replicaManager)

    replay(replicaManager, logManager, quotaManager, partition, log)

    //Create the fetcher thread
    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val leaderEpoch = 1
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread-test1",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = quotaManager,
      brokerTopicStats = null)
    thread.addPartitions(Map(
      t1p0 -> initialFetchState(0L, leaderEpoch),
      t1p1 -> initialFetchState(0L, leaderEpoch)))

    val ResultWithPartitions(fetchRequestOpt, partitionsWithError) = thread.buildFetch(Map(
      t1p0 -> PartitionFetchState(150, None, leaderEpoch, None, state = Fetching, lastFetchedEpoch = None),
      t1p1 -> PartitionFetchState(160, None, leaderEpoch, None, state = Fetching, lastFetchedEpoch = None)))

    assertTrue(fetchRequestOpt.isDefined)
    val fetchRequest = fetchRequestOpt.get.fetchRequest
    assertFalse(fetchRequest.fetchData.isEmpty)
    assertFalse(partitionsWithError.nonEmpty)
    val request = fetchRequest.build()
    assertEquals(0, request.minBytes)
    val fetchInfos = request.fetchData.asScala.toSeq
    assertEquals(1, fetchInfos.length)
    assertEquals(t1p0, fetchInfos.head._1, "Expected fetch request for first partition")
    assertEquals(150, fetchInfos.head._2.fetchOffset)
  }

  @Test
  def shouldFetchNonDelayedAndNonTruncatingReplicas(): Unit = {

    //Setup all dependencies
    val config = KafkaConfig.fromProps(TestUtils.createBrokerConfig(1, "localhost:1234"))
    val quotaManager: ReplicationQuotaManager = createNiceMock(classOf[ReplicationQuotaManager])
    val logManager: LogManager = createMock(classOf[LogManager])
    val log: Log = createNiceMock(classOf[Log])
    val futureLog: Log = createNiceMock(classOf[Log])
    val partition: Partition = createMock(classOf[Partition])
    val replicaManager: ReplicaManager = createMock(classOf[ReplicaManager])

    //Stubs
    val startOffset = 123
    expect(futureLog.logStartOffset).andReturn(startOffset).anyTimes()
    expect(replicaManager.logManager).andReturn(logManager).anyTimes()
    stub(log, null, futureLog, partition, replicaManager)

    replay(replicaManager, logManager, quotaManager, partition, log, futureLog)

    //Create the fetcher thread
    val endPoint = new BrokerEndPoint(0, "localhost", 1000)
    val leaderEpoch = 1
    val thread = new ReplicaAlterLogDirsThread(
      "alter-logs-dirs-thread-test1",
      sourceBroker = endPoint,
      brokerConfig = config,
      failedPartitions = failedPartitions,
      replicaMgr = replicaManager,
      quota = quotaManager,
      brokerTopicStats = null)
    thread.addPartitions(Map(
      t1p0 -> initialFetchState(0L, leaderEpoch),
      t1p1 -> initialFetchState(0L, leaderEpoch)))

    // one partition is ready and one is truncating
    val ResultWithPartitions(fetchRequestOpt, partitionsWithError) = thread.buildFetch(Map(
        t1p0 -> PartitionFetchState(150, None, leaderEpoch, state = Fetching, lastFetchedEpoch = None),
        t1p1 -> PartitionFetchState(160, None, leaderEpoch, state = Truncating, lastFetchedEpoch = None)))

    assertTrue(fetchRequestOpt.isDefined)
    val fetchRequest = fetchRequestOpt.get
    assertFalse(fetchRequest.partitionData.isEmpty)
    assertFalse(partitionsWithError.nonEmpty)
    val fetchInfos = fetchRequest.fetchRequest.build().fetchData.asScala.toSeq
    assertEquals(1, fetchInfos.length)
    assertEquals(t1p0, fetchInfos.head._1, "Expected fetch request for non-truncating partition")
    assertEquals(150, fetchInfos.head._2.fetchOffset)

    // one partition is ready and one is delayed
    val ResultWithPartitions(fetchRequest2Opt, partitionsWithError2) = thread.buildFetch(Map(
        t1p0 -> PartitionFetchState(140, None, leaderEpoch, state = Fetching, lastFetchedEpoch = None),
        t1p1 -> PartitionFetchState(160, None, leaderEpoch, delay = Some(new DelayedItem(5000)), state = Fetching, lastFetchedEpoch = None)))

    assertTrue(fetchRequest2Opt.isDefined)
    val fetchRequest2 = fetchRequest2Opt.get
    assertFalse(fetchRequest2.partitionData.isEmpty)
    assertFalse(partitionsWithError2.nonEmpty)
    val fetchInfos2 = fetchRequest2.fetchRequest.build().fetchData.asScala.toSeq
    assertEquals(1, fetchInfos2.length)
    assertEquals(t1p0, fetchInfos2.head._1, "Expected fetch request for non-delayed partition")
    assertEquals(140, fetchInfos2.head._2.fetchOffset)

    // both partitions are delayed
    val ResultWithPartitions(fetchRequest3Opt, partitionsWithError3) = thread.buildFetch(Map(
        t1p0 -> PartitionFetchState(140, None, leaderEpoch, delay = Some(new DelayedItem(5000)), state = Fetching, lastFetchedEpoch = None),
        t1p1 -> PartitionFetchState(160, None, leaderEpoch, delay = Some(new DelayedItem(5000)), state = Fetching, lastFetchedEpoch = None)))
    assertTrue(fetchRequest3Opt.isEmpty, "Expected no fetch requests since all partitions are delayed")
    assertFalse(partitionsWithError3.nonEmpty)
  }

  def stub(logT1p0: Log, logT1p1: Log, futureLog: Log, partition: Partition,
           replicaManager: ReplicaManager): IExpectationSetters[Option[Partition]] = {
    expect(replicaManager.localLog(t1p0)).andReturn(Some(logT1p0)).anyTimes()
    expect(replicaManager.localLogOrException(t1p0)).andReturn(logT1p0).anyTimes()
    expect(replicaManager.futureLocalLogOrException(t1p0)).andReturn(futureLog).anyTimes()
    expect(replicaManager.futureLogExists(t1p0)).andStubReturn(true)
    expect(replicaManager.onlinePartition(t1p0)).andReturn(Some(partition)).anyTimes()
    expect(replicaManager.localLog(t1p1)).andReturn(Some(logT1p1)).anyTimes()
    expect(replicaManager.localLogOrException(t1p1)).andReturn(logT1p1).anyTimes()
    expect(replicaManager.futureLocalLogOrException(t1p1)).andReturn(futureLog).anyTimes()
    expect(replicaManager.futureLogExists(t1p1)).andStubReturn(true)
    expect(replicaManager.onlinePartition(t1p1)).andReturn(Some(partition)).anyTimes()
  }

  def stubWithFetchMessages(logT1p0: Log, logT1p1: Log, futureLog: Log, partition: Partition, replicaManager: ReplicaManager,
          responseCallback: Capture[Seq[(TopicPartition, FetchPartitionData)] => Unit]): IExpectationSetters[Unit] = {
    stub(logT1p0, logT1p1, futureLog, partition, replicaManager)
    expect(replicaManager.fetchMessages(
      EasyMock.anyLong(),
      EasyMock.anyInt(),
      EasyMock.anyInt(),
      EasyMock.anyInt(),
      EasyMock.anyObject(),
      EasyMock.anyObject(),
      EasyMock.anyObject(),
      EasyMock.capture(responseCallback),
      EasyMock.anyObject(),
      EasyMock.anyObject())
    ).andAnswer(() => responseCallback.getValue.apply(Seq.empty[(TopicPartition, FetchPartitionData)])).anyTimes()
  }
}

redpajama_set_name: RedPajamaGithub


text:
Choose your itinerary
FROM $ 3,372 PP
Trans-European Cruise from Strasbourg to Budapest
13 Days - SBU_AIPP
STRASBOURG - MAINZ - FRANKFURT - WURZBURG - NUREMBERG - REGENSBURG - PASSAU - LINZ - VIENNA - BUDAPEST
From Strasbourg to Budapest, enjoy a unique cruise crossing three rivers and four countries. From the Rhine to the Danube via the Main, you will be dazzled by the beauty of landscapes but also by the sweetness of these rivers that form a link between people, cultures and traditions. France, Germany, Austria and Hungary will unveil their history and rich heritage for an unforgettable trans-European getaway.
Croisi +
All inclusive on board
THE CROISIEUROPE DIFFERENCE
All meals included - DRINKS INCLUDED with meals and at the bar
Refined French cuisine - Gala dinner and evening - Welcome cocktail
Free Wi-Fi in lounge and bar
Headsets are included for excursions
Official welcome from the captain and crew
Travel assistance and repatriation insurance
All port fees included
CRUISE HIGHLIGHTS
A unique cruise combining history, tradition and culture, a passionate encounter
EXCURSIONS INCLUDED:
Frankfurt and its impressive Main Tower
Melk Abbey, a true wonder of Baroque architecture
Vienna and the heritage of the House of Habsburg
Budapest, 2000 years of history
© ONAT Allemagne
Passengers are welcome to board our ship at 6:00 p.m. After comfortably settling into your cabins, we'll introduce our crew at a welcome cocktail. We'll cruise to Mainz.
MAINZ - FRANKFURT
We'll arrive in Mainz early morning. This city, the birthplace of Gutenberg, which lies at the confluence of the Rhine and the Main is a perfect blend of past and present with its thousand year old cathedral. We'll tour the city with your escort before we sail on around in the morning towards Frankfurt. The afternoon, join us for the excursions proposed:
CLASSIC: Sightseeing tour of Frankfurt, a major business hub.
DISCOVERY: Frankfurt "Mainhattan" and the Main Tower: Panoramic tour of Frankfurt and ascent of the Main Tower.
© Grégory Gérault
MILTENBERG - WERTHEIM
We'll stop off in Miltenberg early this morning. Join us to discover this peaceful town with its market place lined with half-timbered houses, its fine main street, hotel Reisen and old Town Hall. We'll have lunch on board before continuing on to Wertheim. Enjoy some free time to discover the town on your own before returning to the ship for dinner.
WERTHEIM - WURZBURG - KITZINGEN
We'll cruise towards Karlstadt. In the afternoon, join us on an excursion for both packages: Visit Wurzburg on the right bank of the Main, which is famous for its baroque palace, former residence of the prince-bishops. In the meantime, the ship cruises on to Wurzburg where we will join it late afternoon. We'll then cruise to Kitzingen. Enjoy an evening of onboard entertaining.
KITZINGEN - ROTHENBURG - BAMBERG
Early this morning we'll start cruising along the Main to reach Schweinfurt. Join us on an excursion for both packages: Guided tour to Rothenburg, a magnificent walled town set up high on the hills over the river, with its narrow streets lined with picturesque half-timbered houses and craftsmen's workshops. It is indeed one of the most beautiful towns on the Romantic Road. In the meantime the ship will cruise on to Schweinfurt where we will join it, before continuing on to Bamberg.
© Scanrail
BAMBERG - NURNBERG - MUHLHAUSEN
This morning will be spent cruising on the Rhine-Main-Danube canal, which was opened on 25th September 1992. We'll have lunch on board before arriving in Nurnberg early afternoon. Join us on an excursion for both packages: Visit of Nuremberg We'll enjoy a panoramic visit of the monuments found outside of the city's medieval walls before arriving in the Marktplatz. On foot, you'll see the gorgeous Schöner Brunnen and the Frauenkirche. During our tour, we'll stop and taste one of the local specialties, Lebkuchen, the city's famous gingerbread. We'll return to the ship during the afternoon and sail on to reach Muhlhausen in the evening. Enjoy an evening of onboard entertainment.
© German National Tourist Board
MUHLHAUSEN - REGENSBURG
This morning will be spent cruising to Kelheim. Join us on an excursion for both packages: Visit of Regensburg, which is the fourth biggest town in Bavaria and also a major industrial hub - with major firms such as BMW, Siemens and Triumph. The historic centre of the city has kept all its original charm with souvenirs of over 20 centuries. In the meantime, the ship cruises on to join us in Regensburg.
© Wolfgang Zwanzger
REGENSBURG - PASSAU
This morning will be spent cruising towards Passau. Enjoy some free time to visit the town on your own. Its baroque buildings and narrow romantic streets are unforgettable. Return to ship and sail on through the attractive region of Engelhartszell. Enjoy an evening of onboard entertainment.
© Joerg Hackemann
PASSAU - MELK - VIENNA
The morning will be spent cruising to reach Melk early in the morning, join us on an excursion for both packages: Visit of the abbey of Melk, perched on a rocky outcrop overlooking the Danube. It is the symbol of Austrian baroque at its finest. We'll return to the ship at the end of the morning and sail on to Dürnstein, arriving there for lunch around noon. Enjoy some free time to explore this picturesque village, and its baroque church, which is the most famous in the Wachau region thanks to the legend of King Richard the Lion Heart. We'll return on board in the afternoon before sailing on to Vienna. Putting in for the night in Vienna. Join us on an excursion for both packages: Tour of the illuminated sights of the city.
VIENNA - ESZTERGOM
Join us for the excursions proposed:
CLASSIC: Visit to the palace of Schöbrunn with its magnificent park.
DISCOVERY: The high art of Viennese living: music and Viennese cafes. Tour of the Konzerthaus concert hall (1).
Enjoy a free afternoon to tour this magnificent city on your own or join us on an excursion proposed: Horse-drawn carriage ride though Vienna. We'll then cruise towards Esztergom, passing Bratislava on the way. Tonight is our dinner and gala evening on board.
ESZTERGOM - BUDAPEST
Enjoy some free time to visit Esztergom on your own, the former capital of Hungary, under the reign of the Arpad kings. Today, it is still a prominent town and the seat of the Hungarian archbishops. Esztergom is decidedly full of treasures with its neo-classical basilica and former Royal palace. We'll set sail once again around noon. Enjoy lunch on board before arriving in Budapest midafternoon. The Hungarian capital which lies along the banks of the Danube, also known as the "pearl of the Danube" is one of the most attractive cities in Europe. Enjoy some free time to explore the city before returning to the ship for dinner. Join us on an excursion for both packages: Visit of Budapest "by night".
© 123 RF
Enjoy a day on your own, or join us for the excursions proposed:
CLASSIC: sightseeing tour of the city.
DISCOVERY: Explore Budapest by bike. Return to the ship. Have a free afternoon to stroll round this attractive city
This afternoon, enjoy some free time or join us on an optional excursion (not included in the packages): Discover Gellert baths and spa. After dinner, enjoy a free evening or join us on an excursion for both packages: traditional folk evening in Budapest.
© Stephane Pfleger
Enjoy one last buffet breakfast on board before disembarking at 9:00 a.m. End of our services.
Choose your departure date
MS Victor Hugo
Ship : MS Victor Hugo Ship category : STANDARD
MS France
Ship : MS France Ship category : STANDARD
MS L'Europe
Ship : MS L'Europe Ship category : STANDARD
Child discount (2-9 years) : - 20%
Your cruise with all meals from dinner on the first day to breakfast on the last day - onboard drinks included (except for drinks from our special lists) - double-occupancy cabin with shower in private bathroom - the classic excursions mentioned in the program - onboard entertainment - assistance from our onboard multi-lingual host/hostess - welcome cocktail - the gala dinner and evening - travel assistance and repatriation insurance - all port fees.
Drinks from our special lists, or drinks ordered during excursions or transfers - cancellation and baggage insurance - personal expenses.
11 Shore excursions
© Österreich Werbung - Viennaslide
© Frantisek Zvardon
Included in tour package Frankfurt Wurtzburg Rothenburg Nuremberg Regensburg Melk Abbey Vienna by night Vienna and the Schoenbrunn Palace Budapest by night Budapest Traditional evening in Budapest
Guided tour of Frankfurt
The excursion to Frankfurt will begin with a panoramic tour of the city by coach. Frankfurt am Main is a dynamic and international city with Germany's most imposing skyline. Under the skyscrapers, you'll find cozy pubs; and at the heart of th...
Guided tour of Würzburg
Departure by coach in the company of your hostess to the center of Würzburg. Our tour will start with a visit to the Würzburg Residence (or Residenz), former palace of the Prince-Bishops of Würzburg. Construction was commissioned by P...
Guided tour of Rothenburg
Departure by coach in the company of your hostess for the charming town of Rothenburg. Located along the "Romantic Road", this town has kept its Renaissance layout virtually intact as it has not undergone any changes since the Thirty Years&...
Guided tour of Nuremberg
Join us for a guided tour of Nuremberg by coach. We'll enjoy a panoramic visit of the monuments found outside of the city's medieval walls, including the Nuremberg Palace of Justice, the Nazi Documentation Centre and Party Rally Grounds, the...
Guided tour of Regensburg
Regensburg saw its political and economic heyday in the 12th and 13th centuries when it was located at a crossroads between major trade routes and became a Free Imperial City in 1245 until other great Bavarian cities were preferred. This poorly indus...
© Zechal
Guided tour of the Melk Abbey
Departure by coach in company of your host/hostess towards Melk Abbey. We will meet up with our guides in the inner courtyard to start our visit. The Abbey overlooking the Danube was first built as the Babenberg's fortified castle. It was founded...
© Austrian National Tourist Office Horvath
Guided tour of Vienna by night
Discover Vienna by night with a panoramic coach tour. This two-thousand-year-old city is a true open-air museum. Vienna embodies Romanticism through its powerful architecture and numerous monuments. We'll take the famous Ring which encircles hist...
Guided tour of Vienna and Schoenbrunn Palace
You will begin the tour by taking the famous Ring to reach Schoenbrunn Palace. The "Ring" was one of the personal projects of Emperor Franz Joseph. In 1857, he decided to take down the ramparts that enclosed the city center and had this bou...
Budapest by night
Explore Budapest by night from the comfort of our coach. Your guide will show you the charms of this capital city with its turbulent but incredibly rich past. Despite the vicissitudes of history, Budapest has managed to preserve and bear witness to a...
© Tomas 1111
Tour of Budapest
In the company of our guides, we will leave for a panoramic tour of Budapest, the Pearl of the Danube. First we will head for the hills of Buda by crossing the Elizabeth Bridge and we will go up to the Castle quarter, from where we'll enjoy a fan...
Traditional evening in Budapest
Transfer by coach in the company of your guides to a theater downtown for a traditional Hungarian show. This one-and-a-half-hour show will take you across all the regions of Hungary through its folklore and Gypsy dance and music. After this show, the...
DISCOVERY EXCURSIONS PACKAGE
© Peter Horvath
Included in tour package Frankfurt and Main tower Wurtzburg Rothenburg Nuremberg Regensburg Melk Abbey Vienna by night The high art of Viennese living: music and Viennese cafes Budapest by night Budapest by bike(1) Traditional evening in...
Frankfurt - "Mainhattan" and the Main Tower
Frankfurt is the largest financial center in continental Europe, earning it the nickname "Mainhatten." During a panoramic tour of the city, you'll be able to admire the huge skyscrapers housing the headquarters of many international ban...
The high art of Viennese living: music and Viennese cafes
Our guide will meet us at the ship, where we'll set off on foot for the subway station to discover the high art of Viennese living. We'll stroll through the city and visit back stage in the Konzerthaus concert hall, which has the Vienna Symph...
© Noppasin
Budapest by bike
From the ship, we'll set out to explore Budapest by bike. You'll see all the important sights of the city on an easy-paced ride, cycling through the historical center of Hungarian culture. We'll pass in front of the basilica, Liberty Squa...
© Bains Gellert
Gellert Baths and Spa
You'll set out with your guide to visit the famous Gellert Baths and Spa. Now part of the famous Hotel Gellert in Buda, references to the spa's healing waters have been found from as early as the 13th century. The current spa was founded in 1...
Horse-drawn carriage ride through Vienna
You'll be transferred to the center of the city to enjoy a horse-drawn carriage ride through the old town and the famous Ring Road. A trip with a Fiaker is a great (and romantic) way of seeing the city in comfort because of the narrow streets and...
Several ships available for this cruise
© Paul Hilbert
STANDARD - 2 Decks
Some administrative formalities to take into account in order to prepare your trip
Valid passport required. Non-EU nationals are advised to consult their embassy or consulate.
For the safety of our passengers, the captain and crew of the boat may decide to modify the navigational itinerary.
(1) The tour of the Konzerthaus might be replaced by a tour of the Musikverein or the opera if it is booked for shows or not available.
Alcohol can be harmful to your health. Please drink in moderation.

redpajama_set_name: RedPajamaCommonCrawl


Download Chhattisgarh PSC Accounting Officer Exam Syllabus 2018 and CGPSC State Service Exam Test Pattern 2018 in PDF format for free of cost. Check CGPSC Excise Tax Inspector Exam Scheme details essential for the candidates to participate in the Chhattisgarh Public Service Commission Written Examination. Get more details about CGPSC SSE Exam Preparation Tips, Question Papers, Exam Study Materials, etc on this page. CGPSC Dy Superintendent Of Police Syllabus 2018 available here in PDF format for applicants reference. Participants who took part in the CGPSC SSE Examination can check the complete details about Chhattisgarh PSC Accounting Officer Syllabus 2018 as well as CGPSC Dy Superintendent Of Police Exam Scheme from here.
Download Chhattisgarh State Service Exam Syllabus 2018 as well as CGPSC Excise Tax Inspector Previous Year Question Papers with Solutions from the links enclosed on this page.
Chhattisgarh Public Service Commission (CGPSC) Officials has released the advertisement to recruit the applicants for Accounting Officer, Dy Superintendent Of Police, Commercial & Excise Tax Inspector & Other Posts. Check the psc.cg.gov.in Dy Superintendent Of Police Syllabus 2018, along with the CGPSC Accounting Officer Exam Pattern, are suggested to follow our page for instant updates on CGPSC Exam Pattern. Participants of Chhattisgarh PSC Accounting Officer Written Examination are advised to stop browsing the CGPSC SSE Exam Scheme. We have disclosed the accurate information about CGPSC State Service Exam Test Pattern 2018 for both Prelims & Mains Exam. Check the below enclosed tabular format for more details about CGPSC Exam Pattern 2018.


text:
Come Inside!
The Candy Counter
The Beal House: Making Dreams Come True
By Katelyn Michaud| June 10, 2019 | Main Street, Food
Something interesting is happening in Littleton. All across northern New England, the median age is increasing as young adults are leaving the rural towns for big city life. But, in Littleton, the median age is actually getting younger. In recent years, an influx of former big-city dwellers are moving to Littleton for a better way of life. And they are opening incredible businesses and giving back to the local community, too.
Lori Alderin and her James Beard-nominated chef husband, Adam, are one of those former big-city dwellers who saw a life-changing opportunity in Littleton and jumped on it. In 2015, Lori, a New Hampshire native, and Adam were living in the southern part of the state. Adam was currently working as an executive chef at a restaurant, but always had a dream of owning his own restaurant someday. Lori had just begun working in real estate when she spotted The Beal House on the market. After seeing the building's historic charm and space, they quickly fell in love with the idea of opening their own restaurant.
The Alderins have been running The Beal House since December 2015. Built in 1833, it wasn't until the turn of the century that the Beal family decided to turn their homestead into an inn. About 25 years ago, the former owners built the restaurant in the old carriage house and barn. Over the years, the various former owners either ran The Beal House as more or as a B&B or a restaurant. The Alderins are currently focused on the restaurant but continue to rent out six comfortable and unique rooms to guests from around the globe.
For a small mountain town, Littleton is home to many delicious restaurants and The Beal House certainly doesn't disappoint the locals or visitors. The restaurant has a rotating menu that uses seasonal locally sourced food whenever possible. Lori drives down to the seacoast to pick up the seafood frequently to ensure that only the freshest ingredients are served to patrons. "We use several local farms for vegetables and proteins when possible. Seasonal availability definitely drives the menu as does access to different products," says Lori.
With a changing menu, it's difficult to determine the most popular dish, but Lori says, "the crispy broccoli is really popular as a starter. Our guests love the scallops as well. Adam does an amazing job with seafood since that's his background." The fried chicken was always a popular choice amongst the locals, so the Alderins turned that into a Sunday night special. In addition to Sunday fried chicken, there's pasta Monday and burger Thursdays.
The food isn't the only delicious things at The Beal House. Many people come just for the cocktails. The menu features many craft cocktails made with house-infused liquors. The drink menu includes numerous national and local craft beers and an extensive wine list for diverse tastes and price point.
The Alderins moved to Littleton primarily to allow Adam to fulfill his lifelong dream of owning his own restaurant after years working as an executive chef. But it has also allowed them to make a lifestyle change. Adam no longer works 10-12 days six days a week. He now works five days a week. While running a business can certainly be hectic at times, Lori and Adam have more time to spend with their daughter. During the months of April and November, the Alderins close the business and take a well-earned vacation.
Both Lori and Adam wanted to be closer to mountains, rivers and lakes, and the ski areas. Littleton was the perfect town for them to settle. "We feel so lucky to have landed in this town. The community and our fellow small business owners have been so incredibly supportive and encouraging. As young entrepreneurs, we couldn't have asked for a more supportive environment or setting," says Lori.
While Littleton may be a bit of a tourism-driven town, it's the locals that keep the Alderin's in business. "We try to feed that flame with weekly specials all year long. Our Thursday Burger Night is like an old home day event sometimes and it can be really fun vibe where everyone knows each other. Everyone is catching up and saying hello," continues Lori.
It doesn't matter if you're a local or just visiting for the night. Head down to The Beal House for delicious locally-sourced food and friendly atmosphere. Who knows, maybe you'll become Littleton's newest residents!
Contact Us/Place Orders
264 Main St. Depot Plaza
Zephyr Quad


Great Taste Magazine 2010 Nov/Dec Issue features Chef Fred Mensinga with Orange County restaurants and professionals with resources, recipes, beautiful photos and many tasty tidbits.
While Southern California struggles to distinguish the seasons with its confusing weather conditions, chefs gather up spices and recipes to remind us that we should start preparing for the holidays. Tis the season!
Start with the tidbits and kudos to discover the good news about OC and neighboring restaurants. Then, check out what Chef Gabriel Caliendo has in The Spice Rack. Chef Caliendo discusses how you can sweeten holiday spirits with cinnamon. Add it to a cup of Chai Tea or to savory dishes.
This month's Cheese Pick, Rogue River Blue Cheese, is smooth, rich, and sweet. With berries, sweet woodsy pine, hazelnuts, and pears, this cheese is another sweet addition to any recipe.
In "Take it or Leave it," Chef Adam Navidi gives an update about the food industry and shares his holiday horror stories about genetically modified foods.
In The Baking Rack, Chef Katie Averill shares her recipe for a gingerbread cake. It is perfect to bring to the family potluck dinner.
As a gift to the community, many restaurants have prepared prix fixe menus that were released before Christmas time. Chef Ahmed Labbate of Charlie Palmer at Bloomingdale's South Coast Plaza talks about this season's restaurant trends in prix fixe menus as well as the advantages and disadvantages of these menus.
For more insight on restaurant trends, Southern California chefs discuss their hopes, speculations, and goals for the restaurant industry in the new year.
Also this month, our featured chef, Chef Fred Mensinga shares his journey through Germany and culinary school, and how he came to be a culinary professional in America.
Finally, Chris Kern, founder of ForgottenGrapes.com and creator of Sunday Night Chef Fights, gets the inside scoop in 2011's wine trends. Keep an eye out and find the great taste in a bottle of wine.
If any of our readers are last-minute Christmas shoppers, have a look at the Holiday Gift Guide. There are many items that will come in handy in the kitchen. 2011, here we come!
Great Taste Magazine 2010 Nov/Dec Issue covers the restaurants and the professionals with resources, recipes, beautiful photos and many tasty tidbits. Great Taste Magazine, the premiere connection to all the key decision makers in our local Southern California restaurant industry. Each month our 15 year-old full-color trade publication reaches thousands of the area's top restaurant owners, general managers, executive chefs and buyers. Take a tour through the magazine. Each issue features a restaurant and local chef, and is packed with the timeliest information, critical operational tips and other industry news important to our readers.


In 2012, a Fire Department in the US responded to a fire somewhere in the nation every 23 seconds. A fire occurred in a structure every 66 seconds, a home fire occurred every 85 seconds. Every 156 seconds, a vehicle caught fire, and every 46 seconds there was a fire in an outside property.
The 1,375,000 fires reported by fire departments in 2012 resulted in an estimated 2,855 civilian deaths and caused an estimate of $12.4 billion in property damage. The average loss per structure fire in 2012 was $20,345.
Are your facility, property and people adequately protected?


text:
Published 6th Apr 2019
Letters to Cozy Glow - Fluttercheer
Cozy Glow
Time after time, Twilight Sparkle tried to find out the reasons for Cozy Glow's actions and got rejected. But maybe a special team of ponies and creatures could succeed where she and her friends have failed. She was not going to give up so easily.
Letters to Cozy Glow written by Fluttercheer
PreviousChapters Next
Prologue: Princess Twilight Sparkle's Friendship Squad2,494 words
A Letter from the Guidance Counselor862 words
A Letter from the Princess of Dreams646 words
To Princess Nightmare Moon889 words
A Letter from a Hurt Friend580 words
A Letter from a Villainous Showmare810 words
A Letter from Three Traitors551 words
A Letter from a Naive Old Stallion809 words
A Letter from the World's Weakest Scholar699 words
A Letter from another Friendship Expert696 words
A Letter from a Sappy Bug541 words
A Letter from Dipcord929 words
A Letter from Commander Tempest709 words
To my Master, Commander Tempest521 words
A Letter with a Useless Friendship Offer574 words
Another Letter from that Persistent Mare510 words
A Letter meant to cheer me up...867 words
To the Cutie Mark Crusaders...16 words
Starlight sent me a letter...412 words
To Starlight...198 words
A Letter from Another Hurt Friend398 words
Starlight, I feel better...208 words
A Dream Conference2,948 words
A Letter about Kites!699 words
I love small talk, Starlight!353 words
A Letter with no Value...904 words
A distraction to make my boredom go away699 words
Thank you, Starlight!185 words
A Letter of Many Colors1,060 words
A Request to my Master, Commander Tempest630 words
The Commander responded!333 words
A Letter to the Brave Changeling King269 words
A Letter to the World's Most Famous Scholar294 words
A Letter to the Real Friendship Expert218 words
A Letter to a Showmare who deserves Forgiveness214 words
A Letter to the Funny Draconequus214 words
A Letter to the Magnificent Wizard from the Past210 words
A Letter to the Night Princess89 words
A Letter to the Three Fillies who are Still my Friends156 words
A Letter to the Persistent Mare...64 words
A Dream Conference
Far away from Cozy Glow and Tartarus, in a place of limitless possibilities, a band of thirteen creatures had gathered. They stood in the middle of a vast expanse of space, stars floating around them in a purple void, and were grouped around a circular stone table, that appeared to be carved out of massive granite. They were all standing, however, there were no chairs for them to sit down on. Unhappiness was plastered on each of their faces, stress and chagrin. The atmosphere was charged, they all felt tense, but the absence of chairs was not the reason for it.
Princess Luna stood at one side of the table, her appearance epic and regal as usual, but her face was adorned with worry, concern and desperation. A proposal of hers had been rejected just now and it made the Princess of the Night anxious as a result. Luna stepped forward and put her forehooves down on the table. It vibrated slightly.
"I must insist!" her voice bellowed, full of strength and demand. "Cozy Glow has suffered nightmares for eight weeks, we cannot allow that her mind will remain unaided like this! Twilight Sparkle's command not to enter her dreams can no longer be honored!"
The crowd sighed and murmured. Twelve faces rested on Luna, faces of resistance and opposition.
"I understand your worries, Princess." Starlight Glimmer took the floor and spoke up. "But I don't think it is a good idea to do this. Cozy trusts us..... starts to trust us, anyway. Entering her dreams would breach that trust. It would be an invasion of her privacy, I can see Twilight's point here. And I think it would also scare her. She already reacted with fear to Moondancer's friendship offer, just imagine how she would react if you would get even closer to her." Starlight's eyes were pleading, urging the princess not to do what she thought would prove to be a very bad idea with terrible consequences.
"I agree," it came from one corner of the table, in a dry, smoky voice. Tempest Shadow stirred and locked eyes with Princess Luna. "I made progress. I have managed to gain the trust of this filly. She listens to me and would do everything I tell her. She even wants to be my student. All the progress would be lost if Cozy Glow shuts us out and refuses to talk because you dug too deep, Princess."
Luna's face shattered and fell apart. They still didn't understand. She turned at her old mentor Starswirl. "At least you will agree, right? You are my teacher and mentor, surely you will see it the way I do, Starswirl?" Luna gave him a cautious smile.
The old wizard remained stoic and calm, but he shook his head. "I am afraid I have to agree with the rest of us, albeit for different reasons. I have heard about the threats the young one spoke out to you and some ponies have more power in their dreams than others. It would be too dangerous to explore the dreams of young Cozy as long as she hates you that much."
"We agree with ya!" A higher voice from the other end of the table, with a slight country drawl, rang out.
At Apple Bloom's side, Scootaloo and Sweetie Belle nodded.
"Yeah, you always helped us, Princess Luna!" Scootaloo remarked.
"And without you, we still wouldn't understand many things," Sweetie Belle added.
"And what's the most important, the sooner ya find out what is wrong with that Cozy Glow, the sooner we don't have ta help her anymore!" Apple Bloom's face became stern and dark.
Once more, Scootaloo and Sweetie Belle showed their support with nods.
Luna's face lit up. But only for a moment, as Stygian spoke up to oppose the three fillies.
"And I disagree," the low voice of the scrawny stallion in his cloak rang out. "Starswirl is right. I have studied dreams and dream magic and how the mind can control and alter a dream. A pony who possesses a lot of anger can do many devastating things in their dreams."
"And we still can't forget Cozy's privacy....." Moondancer chimed in. "I agree with Starlight and Twilight. I don't understand why Cozy rejects me, it is so confusing..... But if Princess Luna enters her dreams without Cozy's agreement, she could lose her trust completely and never talk to any of us ever again." Moondancer's face was filled with grief.
"I think I understand why she rejects you, Moondancer." Starlight Glimmer looked up and over to the other mare. "Cozy is afraid of establishing a real friendship. Her reaction on your letter has shown that clearly. I think she might have been betrayed or abandoned by a friend, like it happened with me and Sunburst."
"Or with me." Tempest Shadow's voice trembled slightly.
Moondancer nodded, her face clearing up a bit. "Or like with me and Twilight. I think this could be the reason..... But would that really be enough to make Cozy reject true friendship completely? I have spoken about her with Twilight, Cozy Glow is so different than me, she is confident and outgoing and never has fears talking to anypony....."
"Heck if I know!" A frustrated, deep voice blurted out before Starlight could respond anything. Gilda threw her claws into the air, then propped her head up on one of them, looking morosely around the crowd. "Now that we know that Cozy Glow is scared of true friendship, this makes everything harder for me! How should I teach her that friendship is a great thing if she is afraid of it? I never had to deal with a griffon who didn't care about friendship because it scared them..... I could use a hint from her dreams." Gilda gave Luna a hopeful glance, a small smile building on her beak.
"Pointless!" Trixie's voice suddenly rang out and the rest of the group looked at her. "It doesn't matter if Princess Luna snoops around in Cozy Glow's dreams or not, it would be a completely pointless thing to do!"
"Why?" Starlight asked, tilting her head and giving her friend a doubtful glance.
"Easy!" Trixie responded with confidence. "Because this filly has lost her mind! Trixie can see that she is completely insane."
Around her, twelve pairs of eyes narrowed in disapproval.
"Don't give Trixie looks like this, Trixie knows what she is talking about!" she shouted, not phased by this reaction. "Trixie was obsessed with power because of the alicorn amulet and the longer Trixie was wearing it, the more insane she became. Cozy Glow is obsessed with power and power obsession turns a pony insane. That's why she had those nightmares, it means Cozy Glow's sanity is getting lost. That's why it's pointless to look into her dreams, you will only find more madness, Princess." Trixie closed her eyes, crossed her forelegs and nodded, to underline her explanation.
"I really don't think that's the case with Cozy..... And it's really mean to say that," Starlight scolded her friend.
Trixie opened her eyes again. "What? Any better ideas?" Trixie shot back and huffed.
"I have a better idea," Thorax responded. His face showed a glare, anger over Trixie's suggestion apparent in it. "Cozy Glow is afraid of making friends because she struggles to decide if she should be good or evil." Thorax' forehead became clad in wrinkles.
"After I left Chrysalis' hive, I was wandering around aimlessly. I stayed hidden and did not dare to approach anypony, not even in disguise. And even after Spike offered me his friendship, I hesitated to take it and he had to convince me first. I was not sure if leaving Chrysalis was really the right decision and everyday I was trying to decide if I should return to the hive. I didn't want to make any friends, I knew it's a pony thing to do and that I would decide for the good side if I did. I wasn't ready to make this decision yet, so I avoided making friends. I'm sure Cozy Glow feels the same way."
"And what do you think about her nightmares?" Luna asked the changeling, bringing the topic back to what was on her heart. "Do you not think I should aid Cozy in her dreams so that she will never have such nightmares again?"
Thorax looked at the princess, his face becoming somewhat softer. "I think they are a good sign. Her nightmares mean that she is close to choosing the good side. She only needs a push into the right direction."
"By entering her dreams," Luna tried it again.
"No," Thorax turned her down. He shook his head. "Cozy Glow sees you as a villain, Princess. If you enter her dreams now, against her agreement, it would give her the feeling that villains can do anything they want without being stopped. It could sway her to the dark side for good."
The confidence in Luna's face faltered and she shrunk a little. "I suppose this makes sense..... I will stay outside of young Cozy's dreams then..... with hesitation."
"Very good! I think that's a splendid idea!" Discord came floating past Luna, a disinterested, bored look on his face, but a mischievous sound in his voice. He had been circling the table the entire time, occasionally plucking some stars off the dream fabric and throwing them into his mouth, where he let his teeth crunch down on them with delight. The draconequus had not paid much attention to the meeting and it was the first time since they had all gathered that he said something.
"Because", he continued, "where would be the fun in entering her dreams and finding out everything all at once? Our audience would be so disappointed if the mystery gets solved that quickly."
"What audience?" Apple Bloom cocked her head and gave him a riddled, confused look.
Discord only flashed a cocky grin, owing her an answer, and kept circling around the table and the gathered creatures.
Starswirl cleared his throat. "Let's bring this conversation into a sane direction again..... I am not very experienced with the troubles that a pony might carry in their mind. But I think it's obvious that the problem of the young one is that she cannot see Moondancer's true intentions. She does not understand that Moondancer wants to help, because she fails to slip into Moondancer's shoes."
Not far away from Starswirl, Stygian's face brightened, the outward signs of a transpiring epiphany. "This must be the solution....." he breathed.
Instantly, all eyes were on him, weary eyes that craved an end of this meeting.
"Cozy Glow could not see it from Moondancer's point of view and that's why she misunderstood her actions. If a friend betrayed her and that's why she is afraid of real friendships, then maybe this friend of hers did not really betray her, maybe they intended to do something else and Cozy Glow just did not understand because she failed to see it from her friend's view!" Stygian put his forehooves on the table and rose up, excited because of his realization.
"Oh, really?" Apple Bloom asked, a snark in her voice. "Ya mean like Cozy failed ta see it from our point of view?" She gestured with her hoof at herself and each of her two friends.
"Um-hm." Scootaloo and Sweetie Belle nodded together, their faces dark and serious.
"Yes," Stygian confirmed. "Cozy Glow did not understand that you didn't betray her friendship, that you just wanted to protect your kingdom and home."
"Great, so Cozy went and betrayed us instead....." Scootaloo dropped her head and sighed, eyes closed.
"Not necessarily," Stygian suggested.
"Huh?" Apple Bloom raised an eyebrow. "Ya ain't serious, are ya?"
"I am. It might not have been Cozy's intent to betray your friendship." Stygian resisted the glare of the filly undeterred.
"That's..... Yeah, sorry, but that's completely ridiculous," Sweetie Belle judged the suggestion, bluntly.
"Yeah, Cozy Glow is just a rotten, no-good filly, of course she betrayed us!" Apple Bloom huffed, her face turning slightly red.
Stygian did not falter. "When I took the artifacts of my friends and wanted to replicate them, I did not try to betray them. I just wanted to find a way to be as strong as them and to fight at their side. But they didn't understand what I was trying to do, they could not see it from my point of view, and so they thought that I am a villain who turns against them." Stygian increased the intensity of his stare at the three fillies. "Maybe Cozy Glow was the same. Maybe she had very important reasons for what she did, reasons that you don't understand, and that is why she had to stop you. I never tried to betray my friends, maybe Cozy Glow did not betray you, either. She could not see your actions from your point of view, but neither do you try to see her actions from her own point of view."
Apple Bloom raised a hoof and opened her mouth to say something, then her angry expression changed to one of surprise. She took her hoof down again, eyelids lowering, then looked down at the table. Scootaloo and Sweetie Belle followed her example. All three of them stayed silent.
"So, are we going with this?" Starlight Glimmer took the floor as Stygian did not say more. "That Cozy Glow feels betrayed by a friend, but that this friend of hers did not really betray her and Cozy just believes they did because she didn't put herself into her friend's shoes? Are we taking this approach in our next letters? I think it is a good theory about Cozy's reason and that it's worth a try, but what do all of you think?" Starlight let her eyes glide over the gathered ponies and creatures.
Tempest Shadow huffed and her answer was a growl, coming out between her gritted teeth.
"I don't agree with it. I think Cozy Glow has a very different reason for acting like this," Thorax responded.
Starswirl and Stygian nodded, the former with closed eyes, the latter with a stern look on his face, but both of them silently expressing their agreement.
"Trixie does not agree. Cozy Glow is just insane and needs a different kind of treatment."
"I am on Starlight's side," Moondancer announced. "I felt so betrayed by Twilight and I think Cozy goes just through the same thing."
"Oh, do whatever you think is right to make that dark spot in Cozy Glow's heart go away, I don't care." Discord continued munching away at the stars.
Apple Bloom, Scootaloo and Sweetie Belle did not answer Starlight's question. They just eyed the table, sunken in thoughts.
"I'll take it!" Gilda spoke gruffly. "Anything that gives me a hint what I should do with that filly now."
"I agree, as well," Luna said as the last one. "I cannot enter young Cozy's dreams to find out what is wrong, but I am willing to try out any idea that offers itself."
Starlight looked at each pony and creature as they cast their vote on the subject. Then she nodded. "Okay, so we have six creatures who agree with the new plan, three creatures who disagree with it and four abstentions. That settles it. We will try to make Cozy realize that her friend from the past did not intend to betray her, to fix her fear of true friendship."
Six creatures nodded, while Tempest Shadow let out a longer and sharper growl than before, neither the sound nor the expression in her face doing any effort to mask how much she detested Stygian's theory.
"Now we only need to decide who will start this new approach and will send Cozy the next letter," Starlight continued. "Princess Luna and Moondancer are unfortunately being ruled out, this leaves only myself, Starswirl, Stygian and Gilda. And if none of you mind, I think I should be the one who begins and send Cozy another letter. We can't just simply tell Cozy that her friend did not mean to betray her, if what we think is true, she will be a hurt a lot and reject us if we are too direct about it. We need somepony who will gently dig for information, without upsetting Cozy."
"Go ahead," Gilda prompted her, casually waving a claw at Starlight. "I never had to teach friendship to someone with a problem like this, I need more infos before I can figure out what to do."
"I think the choice is excellent," Starswirl spoke confidently. "I can teach the young one to see things from the perspective of her friend once you have succeeded to find out what happened in her past." The old wizard stroke over his beard.
"I agree," Stygian simply said.
"Okay. I'm glad you're seeing it my way." Starlight gave the griffon and the two ancient ponies a nod each. "I guess that means we are done here. Princess Luna?" She turned at the princess.
"It seems you are correct, Starlight. This dream meeting is hereby over. I will set you all free of this shared dream now and let you back into your individual dreams. Rest well, ponies and creatures, for we have an important mission to accomplish."
Most creatures of the crowd nodded, then they disappeared one after another, until only Luna and Discord were left. The draconequus continued his business, not paying heed to the princess.
Luna did not pay heed to him, either. She fixated the table, from under sunken eyelids. "I hope I am doing the right thing, young Cozy Glow."
I am very proud of this chapter. The words were flowing so good like they haven't in a long time. My writing has slowly returned back to old form since I picked up the pen again in December, with more and more improvements being made. Writing this chapter was only the second time that I could write so freely again since December. This chapter has turned out much better than most of my other work in the last eight months.
I am even so proud of this chapter, that I dare to plug my Patreon and Ko-fi account again for the first time in a long while.
If you like this chapter as much as I do, consider becoming my patron or giving me a little tip on Ko-fi:
https://www.patreon.com/fluttercheer
https://ko-fi.com/fluttercheer
Bronycommander
Bronycommander	#1 · Aug 7th, 2020 · 1 · ·
Great job, I loved it!
Kichi
Kichi	#2 · Aug 7th, 2020 · 1 · ·
Maybe before that, they should try to search about Cozy past, as it seems they don't really know much about her
Blast	#3 · Aug 9th, 2020 · 1 · ·
Gilda knows her chances.
Fluttercheer
Fluttercheer	#4 · Aug 10th, 2020 · · ·
I'm sure that would be a good idea.
Bronycommander	#5 · Aug 10th, 2020 · · ·
You're Welcome. And i agree, the Show Never did that, which i find was a mistake. I mean, What Brings a filly to do something as this
Yosh-E-O
Yosh-E-O	#6 · Aug 11th, 2020 · 1 · ·
You certainly have all right to feel proud of this chapter in the story. It is not just an advancement in the series, but also one that helps to recap what has been achieved so far.
The differing opinions on what should be the proper way to help Cozy made this a most fascinating chapter. One would expect that Luna would push her way until everyone agreed. Instead, we see a genuine discussion with different emotions playing a role in what would make for the best way to press forward.
What is also good is how, thanks to Stidgen, , we get a not so often, if ever, pursued route of seeing cozy's behavior as being a result of feeling betrayed while not having the ability to see from someone else's point of view. It's a great combination of what Stidgen and Starlight experienced.
You can also easily understand Tempest Shadow's reason for not really saying anything after the meeting was nearing its end. I'm sure she still has a hard time trying to meet with Glitter Drops and Spring Rain after how things went before she left for her horrible times alone and into the waiting, manipulative hands of The Storm King.
The value in this chapter is getting canon points of view while also recalling what has been and working towards what will be. Seeing a non Luna gets her way situation play out is a nice twist as, again, one would think the chapter would guide to her solution being the only right solution.
Lots to take away here. Great job!


text:
Anne Frank: Diary Of A Young Girl Discussion Guide
Use these enrichment activities and Internet resources to enhance teaching Anne Frank: Diary of a Young Girl.
Literature (1,453)
Banned Books and Censorship (63)
Reading and Literature (3,642)
Journaling Tips, Strategies & Topics (32)
World War II (1939-1945) (56)
Social Studies and History (3,353)
Letters, Journals and Diaries (59)
Historic Wars and Military Action (73)
by TeacherVision Staff
Updated on: February 1, 2002
Anne Frank: Diary of a Young Girl
by Anne Frank
Born on June 12, 1929, Anne Frank was a Jewish girl and only 13 years old when she was forced to go into hiding during the Holocaust. She and her family, along with four others, spent 25 months during World War II in an annex of rooms above her father's office in Amsterdam, the Netherlands, before they were found by the Nazis and sent to concentration camps. Although Anne did not survive, her diary did. It was saved during the war by one of the family's helpers, Miep Gies, and was first published in 1947. Through this diary we see Anne's life, set against a background of unbelievably horrific world events.
To buy this book, click here or on the book cover.
The Diary of Anne Frank Literature & Film Guide
Pair your study of the classic book The Diary of Anne Frank with the PBS MASTERPIECE Classic film version of the story. This guide to both the book and film will extend students' understanding of the story and the era in which it was written.
Acrostic Poem
Write an acrostic poem that describes Anne Frank's character. Use her diary as a reference.
The Anne Frank and Miep Gies Connection
After reading a Dateline report on Miep Gies, who was a friend to Anne Frank's family, students answer questions about Miep's life.
Culminating Projects
Students choose from a selection of five motivating and thought-provoking projects. Projects include extending Anne's diary entries, analyzing her Gemini personality traits, and completing "what if" scenarios. Use the Anne Frank: Diary of a Young Girl Project Ideas sheet.
Draw a Floor Plan
Have students draw a floor plan of what they think the annex looked like, based on Anne's descriptions in her diary.
Holocaust Remembrance Day (Yom HaShoah)
Explore a collection of resources for Holocaust Remembrance Day, which takes place on April 11, 2010.
TeacherVision's World War II Theme
Check out the lessons in our World War II Theme. Activities and resources include a timeline of the war, a photographic images lesson, and more.
"What If" Creative Writing
Give students the following question to answer: If you had to go into hiding for an indefinite period of time, what objects or materials would you take with you? Have them write on the blank journal page and then display their writing for others to see.
Comprehension Checks
Check your students' comprehension of Anne Frank: Diary of a Young Girl with various tests.
Curious George Authors: The Reys
Share the story of the Reys, creators of Curious George books, and assign the related activities to learn more about these... Read more
Writers of the Forties
Learn about three major writers of the forties: Richard Wright, James Michener, and Herman Wouk. Read more
Samuel Pepys - Biography & Activities
Writer ... Read more
Emily Dickinson -- Selected Poems -- Series One
Emily Dickinson – Selected Poems – Series One Note: Of the 1,775 poems Emily Dickinson (1830-1886) wrote, only a very few... Read more
Emily Dickinson -- Selected Poems -- Series Two
Emily Dickinson – Selected Poems – Series Two Note: Of the 1,775 poems Emily Dickinson (1830-1886) wrote, only a very few were... Read more
Homesick: My Own Story
Homesick: My Own Story ... Read more


text:
You are at:Home»Competition»FIN reveal Pellegrini received €750,000 between 2006 and 2012
FIN reveal Pellegrini received €750,000 between 2006 and 2012
Read La Gazetta Dello Sport
Following the champion's scathing attack on twitter over the 2009 World Championships prize money, the Italian swimming federation, FIN have responded with a depth charge and made public the earnings of the Italian swimmer between 2006 and 2012. "The athlete Federica Pellegrini has personally received a total of around 750.000 euro, for her ranking in European competitions and study grants; this excludes prizes relating to the Olympic Club and the contributions from CONI for the Olympic medals."
147,000 euro for the Rome championships alone.
Fede hits back at the FIN: "I am shocked at the statements made by the president of my sport. It would be all to easy to respond to the mathematical calculation of the prizes by looking at the long list of medals and records achieved between 2004 and today. And all of this becaause I criticised them."
Image courtesy of Wikimedia Commons, CC BY 2.0
62.091774-6.747201
Federica Pellegrini


text:
Reeves v. Sanderson Plumbing Prods. Inc.
Roger Reeves (P) was 57 years old and had spent 40 years in the employ of Sanderson Plumbing Products, Inc. (D), a manufacturer of toilet seats and covers. P's responsibilities included recording the attendance and hours of those under his supervision and reviewing a weekly report that listed the hours worked by each employee. Management was informed that 'production was down' in P's area because employees were often absent and were 'coming in late and leaving early.' Monthly attendance reports did not indicate a problem. An audit was ordered and 'numerous timekeeping errors and misrepresentations on the part of Caldwell, Reeves, and Oswalt were discovered.' P and Caldwell were fired. P sued contending that he had been fired because of his age in violation of the Age Discrimination in Employment Act of 1967. D contended that it had fired P due to his failure to maintain accurate attendance records, while P attempted to demonstrate that D's explanation was pretext for age discrimination. The District Court twice denied oral motions by D for judgment as a matter of law under Rule 50 of the Federal Rules of Civil Procedure, and the case went to the jury. The court instructed the jury that 'if the plaintiff fails to prove age was a determinative or motivating factor in the decision to terminate him, then your verdict shall be for the defendant.' P got the verdict and $35,000 in compensatory damages, and found that D's age discrimination had been 'willful.' Judgment was entered for $70,000. D then renewed its motion for judgment as a matter of law and alternatively moved for a new trial. The District Court denied D's motions and granted P's, awarding him $28,490.80 in front pay for two years' lost income. The Court of Appeals for the Fifth Circuit reversed, holding that petitioner had not introduced sufficient evidence to sustain the jury's finding of unlawful discrimination. The court acknowledged that petitioner 'very well may' have offered sufficient evidence for 'a reasonable jury [to] have found that [respondent's] explanation for its employment decision was pretextual.' But, that was 'not dispositive' of the ultimate issue--namely, 'whether Reeves presented sufficient evidence that his age motivated [respondent's] employment decision.' The court noted that age-based comments 'were not made in the direct context of Reeves's termination.'


No figure of Christ can be more beautiful than that of a shepherd, His tender care, devotedness, faithfulness, meekness, patience and sympathy.
Joseph rises on the horizon of his time like a beautiful star, that shines on and sets without a single cloud to dim its brightness. He is the Sirius of the shining host of Old Testament typical men.


text:
Home > College Football
FBSAACACCBIG 12BIG TENCUSAINDMACMWCPAC 12SBCSEC
CUSA
1 LSU 1,535 13 0 0
2 Ohio State 1,488 13 0 0
3 Clemson 1,441 13 0 0
4 Oklahoma 1,364 12 1 0
5 Georgia 1,241 11 2 0
6 Florida 1,167 10 2 0
7 Oregon 1,141 11 2 0
8 Baylor 1,039 11 2 0
9 Alabama 1,011 10 2 0
9 Auburn 1,011 9 3 0
Coaches Poll Top 10
10 Utah 1,005 11 2 0
FBS Top 25 Scoreboard
Las Vegas Bowl
(18) Boise State (12-2) 0 0 7 0 7
Washington (8-5) 7 10 7 14 38
H. Bachmeier (BSU), 26-15-119, 2 INT, 0 TD
J. Eason (WASH), 32-22-210, 0 INT, 1 TD
G. Holani (BSU), 11-35
R. Newton (WASH), 15-69, 1 TD
K. Shakir (BSU), 3-38
T. Bynum (WASH), 5-67, 1 TD
(20) Appalachian State (13-1) 0 10 21 0 31
UAB (9-5) 14 0 3 0 17
Z. Thomas (APST), 24-13-142, 0 INT, 2 TD
T. Johnston (UAB), 34-22-298, 1 INT, 2 TD
D. Evans (APST), 19-161, 1 TD
S. Brown (UAB), 14-39
J. Virgil (APST), 3-62
A. Watkins (UAB), 10-159, 1 TD
(25) Oklahoma State (8-5) 14 0 0 7 21
Texas A&M (8-5) 0 7 7 10 24
D. Brown (OKST), 28-15-184, 0 INT, 2 TD
K. Mond (TAMU), 19-13-95, 0 INT, 1 TD
C. Hubbard (OKST), 19-158
K. Mond (TAMU), 12-117, 1 TD
B. Johnson (OKST), 5-124, 2 TD
Q. Davis (TAMU), 6-48
(22) USC (8-5) 7 10 7 0 24
(19) Iowa (10-3) 7 21 7 14 49
K. Slovis (USC), 30-22-260, 0 INT, 2 TD
N. Stanley (IOWA), 27-18-213, 0 INT, 2 TD
V. Malepeai (USC), 8-37
T. Goodson (IOWA), 18-48, 1 TD
A. St. Brown (USC), 9-163
N. Ragaini (IOWA), 2-54
Cheez-It Bowl
(24) Air Force (11-2) 0 17 7 7 31
Washington State (6-7) 0 14 0 7 21
D. Hammond (AF), 12-4-30, 1 INT, 0 TD
A. Gordon (WSU), 42-28-351, 0 INT, 3 TD
K. Remsberg (AF), 26-178, 1 TD
M. Borghi (WSU), 6-27
B. Waters (AF), 2-12
B. Arconado (WSU), 11-167, 1 TD
Camping World Bowl
(14) Notre Dame (11-2) 10 10 10 3 33
Iowa State (7-6) 0 6 3 0 9
I. Book (ND), 28-20-247, 0 INT, 1 TD
B. Purdy (IAST), 30-17-222, 0 INT, 0 TD
T. Jones (ND), 11-135, 1 TD
B. Hall (IAST), 17-55
C. Claypool (ND), 7-146, 1 TD
L. Pettway (IAST), 4-54
(15) Memphis (12-2) 13 10 13 3 39
(13) Penn State (11-2) 7 28 10 8 53
B. White (MEM), 51-32-454, 2 INT, 0 TD
S. Clifford (PSU), 20-11-133, 1 INT, 1 TD
P. Taylor (MEM), 8-50, 1 TD
J. Brown (PSU), 16-202, 2 TD
D. Coxie (MEM), 8-132
K. Hamler (PSU), 2-46
(4) Oklahoma (12-2) 7 7 7 7 28
(1) LSU (14-0) 21 28 7 7 63
J. Hurts (OKLA), 31-15-217, 1 INT, 0 TD
J. Burrow (LSU), 39-29-493, 0 INT, 7 TD
J. Hurts (OKLA), 14-43, 2 TD
C. Curry (LSU), 16-89
C. Lamb (OKLA), 4-119
J. Jefferson (LSU), 14-227, 4 TD
(3) Clemson (14-0) 0 14 7 8 29
(2) Ohio State (13-1) 10 6 0 7 23
T. Lawrence (CLEM), 33-18-259, 0 INT, 2 TD
J. Fields (OSU), 46-30-320, 2 INT, 1 TD
T. Lawrence (CLEM), 16-107, 1 TD
J. Dobbins (OSU), 18-174, 1 TD
T. Etienne (CLEM), 3-98, 2 TD
K. Hill (OSU), 6-67
(6) Florida (11-2) 14 10 3 9 36
Virginia (9-5) 7 7 0 14 28
K. Trask (FLA), 39-24-305, 1 INT, 1 TD
B. Perkins (UVA), 40-28-323, 1 INT, 4 TD
L. Perine (FLA), 13-138, 2 TD
B. Perkins (UVA), 14-24
V. Jefferson (FLA), 6-129
T. Jana (UVA), 7-126, 1 TD
Liberty Bowl
(21) Navy (11-2) 3 7 7 3 20
Kansas State (8-5) 0 10 0 7 17
M. Perry (NAVY), 7-5-57, 0 INT, 1 TD
S. Thompson (KSST), 14-10-124, 0 INT, 0 TD
M. Perry (NAVY), 28-213
J. Gilbert (KSST), 8-39
C. Warren (NAVY), 1-41
W. Gill (KSST), 2-57
(12) Utah (11-3) 0 0 3 7 10
Texas (8-5) 3 7 14 14 38
T. Huntley (UTAH), 23-15-126, 0 INT, 1 TD
S. Ehlinger (TEX), 18-12-201, 1 INT, 3 TD
Z. Moss (UTAH), 16-57
K. Ingram (TEX), 13-108, 1 TD
C. Fotheringham (UTAH), 3-30
D. Duvernay (TEX), 3-92, 1 TD
(17) Michigan (9-4) 10 6 0 0 16
(9) Alabama (11-2) 7 7 7 14 35
S. Patterson (MICH), 37-17-233, 2 INT, 1 TD
M. Jones (ALA), 25-16-327, 0 INT, 3 TD
Z. Charbonnet (MICH), 13-84
N. Harris (ALA), 24-136, 2 TD
G. Jackson (MICH), 4-57
J. Jeudy (ALA), 6-204, 1 TD
(16) Minnesota (11-2) 10 14 0 7 31
(9) Auburn (9-4) 10 7 7 0 24
T. Morgan (MINN), 29-19-278, 1 INT, 2 TD
B. Nix (AUB), 26-17-176, 0 INT, 1 TD
M. Ibrahim (MINN), 20-140, 1 TD
J. Whitlow (AUB), 9-24, 1 TD
T. Johnson (MINN), 12-204, 2 TD
A. Schwartz (AUB), 6-49
(7) Oregon (12-2) 7 7 7 7 28
(11) Wisconsin (10-4) 10 7 7 3 27
J. Herbert (ORE), 20-14-138, 1 INT, 0 TD
J. Coan (WISC), 35-23-186, 1 INT, 1 TD
C. Verdell (ORE), 17-49
J. Taylor (WISC), 21-94
J. Johnson (ORE), 5-66
Q. Cephus (WISC), 7-59, 1 TD
(5) Georgia (12-2) 3 16 7 0 26
(8) Baylor (11-3) 0 0 14 0 14
J. Fromm (UGA), 30-20-250, 0 INT, 2 TD
C. Brewer (BAY), 41-24-211, 1 INT, 1 TD
Z. White (UGA), 18-92, 1 TD
T. Ebner (BAY), 5-23
G. Pickens (UGA), 12-175, 1 TD
D. Mims (BAY), 5-75, 1 TD
Thu 1/2
Boston College (6-7) 0 0 6 0 6
(23) Cincinnati (11-3) 7 10 7 14 38
D. Grosel (BC), 17-8-87, 0 INT, 0 TD
D. Ridder (CINN), 24-14-95, 0 INT, 1 TD
D. Bailey (BC), 8-28
D. Ridder (CINN), 21-105, 3 TD
H. Long (BC), 2-45
J. Jackson (CINN), 2-33
CFP National Championship Game
(1) LSU (15-0) 7 21 7 7 42
T. Etienne (CLEM), 15-78, 1 TD
C. Edwards-Helaire (LSU), 16-110
J. Ross (CLEM), 5-76
J. Chase (LSU), 9-221, 2 TD
FBS Leaders
Passing Yardage
Joe BurrowJ. Burrow LSU 5,671
Anthony GordonA. Gordon WSU 5,579
Cole McDonaldC. McDonald HAW 4,135
Brady WhiteB. White MEM 4,014
Brock PurdyB. Purdy IAST 3,982
Rushing Yardage
Chuba HubbardC. Hubbard OKST 2,094
Malcolm PerryM. Perry NAVY 2,017
J.K. DobbinsJ. Dobbins OSU 2,003
Jonathan TaylorJ. Taylor WISC 2,003
Jaret PattersonJ. Patterson BUFF 1,799
Receiving Yardage
Ja'Marr ChaseJ. Chase LSU 1,780
Omar BaylessO. Bayless ARST 1,661
Justin JeffersonJ. Jefferson LSU 1,540
Antonio Gandy-GoldenA. Gandy-Golden LIB 1,396
Devin DuvernayD. Duvernay TEX 1,386
A Twitter List by TheFootballDB