TL/DR: How can I start a KafkaSource in Flink 1.15.2 from manually provided offsets when starting from a snapshot that contains committed offsets for the same consumer group? Hi Team, I am building a multi-region streaming pipeline using AWS MSK as the messaging backbone and Flink processors running on kinesis analytics (version 1.15.2, since that is the latest that kinesis analytics supports). In order to support our disaster recovery approach, we have a need to override the offsets for our KafkaSources to those of another consumer group (the same processor in the other region). We use MM2 on MSK connect for our data replication, which also translated and replicates consumer group offsets. When trying to override the starting offsets for one of our sources, I am noticing that the kafka source is starting from the committed offsets that have been persisted in the state snapshot rather than the offsets being provided by our override logic. I have logs showing that the offset initializer contains the correct offsets for each topic partition, but then the logs created by the kafka source on startup instead use the offsets that have been stored in the snapshot. Is it not an option for us to start without a snapshot, since our pipelines are stateful and require the ability to access previously processed data. Is there a way to force flink to use the offsets we are providing as part of the override, instead of preferring what is stored in state? Thank you in advance for your help. For reference, this is the code that returns the offset initializer and logs the intended offsets:

private def overrideOffsets(topicName: String)(implicit config: BaseKafkaConfig): OffsetsInitializer = {
    val adminClient = AdminClient.create(config.createKafkaBasePropMap.toProperties)

    val consumerGroupName =
      s"${JarvisKafkaConstants.CONSUMER_GROUP_PREFIX}.${config.getAlternateRegionDesignation}.${config.getProcessorName}"
    <http://logger.info|logger.info>(s"Getting offsets for consumer group: $consumerGroupName")

    val replicatedTopicOffsetMap = adminClient.listConsumerGroupOffsets(consumerGroupName)
    <http://logger.info|logger.info>(s"Retrieved offset map: ${replicatedTopicOffsetMap.partitionsToOffsetAndMetadata().get().toString}")

    val offsetMap: collection.Map[TopicPartition, java.lang.Long] = replicatedTopicOffsetMap
      .partitionsToOffsetAndMetadata()
      .get()
      .asScala
      .filterKeys(partition => partition.topic() == topicName)
      .map { case (partition, offsetAndMetadata) => partition -> long2Long(offsetAndMetadata.offset()) }

    <http://logger.info|logger.info>(offsetMap.foldLeft(s"Starting consumer with the following overridden offsets for topic $topicName.") {
      case (s: String, (partition, offset)) => s + s"\n${partition.topic()}, ${partition.partition()} -> $offset"
    })

    OffsetsInitializer.offsets(offsetMap.asJava)
  }

And this is the code that creates the kafka source, where the offset initializer is passed in as an input argument:

private def createKafkaSourceWithOffsets(
    topicName: String,
    startingOffsets: OffsetsInitializer
  )(implicit config: BaseKafkaConfig): KafkaSource[String] =
    KafkaSource
      .builder[String]()
      .setTopics(topicName)
      .setStartingOffsets(startingOffsets)
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .setProperties(config.createKafkaConsumerProperties)
      .build()

Flink doesn’t rely on committed offsets for fault tolerance, everything is stored in Flink snapshots. You can’t force Flink to use the provided offsets. The only possible option is to fork Flink, figure out where Flink does this internally and use that forked version. The reason for not being able to override it, is because of the processing guarantees

Consumer groups for the same application in each region actually are different in our pipeline, e.g. primary.<group_name> and secondary.<group_name>. There were a lot of factors that played into this decision. Each region has a set of native topics and replicated topics (via MM2) from the other region, and each processor consumes from both topics. The intention is: one data source in one region, but processors in both regions process the same data and build their internal state.

Depending on your job graph, changing the source offsets alone might not be a great idea. For instance, if you have downstream stateful operators such as windows, or are using event time. A better solution might be to just restart the app without state, and change consumer group/set the connector to use the committed offsets:

.setStartingOffsets(OffsetsInitializer.committedOffsets())

. If you cannot change the consumer group then you could use a timestamp

.setStartingOffsets(OffsetsInitializer.timestamp(1657256176000L))

In KDA you could externalise the consumer group/timestamp to runtime properties and read when constructing the job graph.

Yeah, unfortunately we do need to preserve the state in failover scenarios. But I appreciate the recommendation.

No it's not possible as @Martijn Visser said without forking the connector

That’s exactly right! We have a flag that we set in our application config properties only during the first restart after a failover.

Good luck!

Wanted to share a quick update here: Our updated approach to make sure that we are able to process records that happen during a regional impairment or failover is to leverage our existing snapshot manager and just restore our jobs to their latest snapshots that were taken before the time of impairment. While this may lead to some reprocessing of older records, we have mechanisms built into our processors to handle this and will consider the reprocessing time part of our RTO. Thanks for the help everyone!