class FanOutRecordPublisher implements RecordPublisher {

	@Override
	public RecordPublisherRunResult run(final RecordBatchConsumer recordConsumer) throws InterruptedException {
		LOG.info("Running fan out record publisher on {}::{} from {} - {}",
			subscribedShard.getStreamName(),
			subscribedShard.getShard().getShardId(),
			nextStartingPosition.getShardIteratorType(),
			nextStartingPosition.getStartingMarker());

		Consumer<SubscribeToShardEvent> eventConsumer = event -> {
			RecordBatch recordBatch = new RecordBatch(toSdkV1Records(event.records()), subscribedShard, event.millisBehindLatest());
			SequenceNumber sequenceNumber = recordConsumer.accept(recordBatch);
			nextStartingPosition = StartingPosition.continueFromSequenceNumber(sequenceNumber);
		};

		RecordPublisherRunResult result = runWithBackoff(eventConsumer);

		LOG.info("Subscription expired {}::{}, with status {}",
			subscribedShard.getStreamName(),
			subscribedShard.getShard().getShardId(),
			result);

		return result;
	}

	public  FanOutRecordPublisher(
			final StartingPosition startingPosition,
			final String consumerArn,
			final StreamShardHandle subscribedShard,
			final KinesisProxyV2Interface kinesisProxy,
			final FanOutRecordPublisherConfiguration configuration,
			final FullJitterBackoff backoff);

}