title: Kafka Scala: How to get offset value from OffsetAndMetadata class
body: <p>I'm looping thrice on <code>offsets</code>:</p>
  <ol>
  <li>Get <code>TopicPartition</code> &amp; <code>offset</code></li>
  <li>Get <code>TopicPartition</code> &amp; <code>OffsetAndMetadata</code></li>
  <li>Get the delta between the producer &amp; consumer</li>
  </ol>
  <p>I'm wondering if I can get the <code>offset</code> value from <code>OffsetAndMetadata</code> but I'm not sure how. I couldn't find examples online to get this value. Any help is appreciated, thank you!</p>
  <pre><code>override def onQueryProgress(event: StreamingQueryListener.QueryProgressEvent): Unit = {
      event.progress.sources
        // Ignoring sqs / jms sources their offsets
        .filter(source =&gt; source.description.toLowerCase().contains(&quot;kafka&quot;))
        // ex offset :
        //    &quot;endOffset&quot; : {
        //      &quot;rcs-enriched-event&quot; : {
        //        &quot;8&quot; : 31376,
        //        &quot;11&quot; : 39114,
        //        &quot;2&quot; : 39376,
        //      } ...
        .foreach(source =&gt; {
  
          /// Map[Topic,Map[Partition, CurrentOffset]]
          val jsonOffsets = objectMapper.readValue(source.endOffset, classOf[Map[String, Map[String, Int]]])
          jsonOffsets.keys.filter(key =&gt; topics.contains(key))
            .foreach(topic =&gt; {
              val offsets: Map[String, Int] = jsonOffsets(topic)
              val consumedPartitions = new ListBuffer[TopicPartition]()
              val topicOffsetList = new ListBuffer[Int]()
  
              val mapTopicPartitionOffset = offsets
                .keys
                .map(partition =&gt; {
                  val tp = new TopicPartition(topic, partition.toInt)
                  val offset = offsets(partition).toLong
                  (tp -&gt; offset)
                })
                .toMap
  
              val mapTopicPartition = offsets
                .keys
                .map(partition =&gt; {
                  val tp = new TopicPartition(topic, partition.toInt)
                  val oam = new OffsetAndMetadata(offsets(partition).toLong)
                  (tp -&gt; oam)
                })
                .toMap
  
  
              for(topicPartition &lt;- mapTopicPartitionOffset){
                consumedPartitions += topicPartition
              }
  
              try {
                val kafkaPartitionOffset = kafkaConsumer.endOffsets(consumedPartitions.asJava)
  
                for((topicPartition,offset) &lt;- mapTopicPartitionOffset){
  
                  val bbCurrentOffset =  offset
  
                  // latest offset
                  val partitionLatestOffset = kafkaPartitionOffset.get(topicPartition)
  
                  // Partition offset delta
                  val delta = partitionLatestOffset - bbCurrentOffset
  
                  topicOffsetList += delta.abs
                }
              } catch {
                case e: Exception =&gt; {
                  log.error(s&quot;${consumerGroupId} Could not get Kafka offset&quot;, e)
                }
              }
  
              try {
                kafkaConsumer.commitSync(mapTopicPartition.asJava)
              } catch {
                case e: Exception =&gt; log.error(s&quot;${consumerGroupId} Could not commit offset&quot;, e)
              }
  
              //log.info have the group id (unique id), the topic, cumulative consumer lag delta
              log.info(&quot;consumerGroupId: &quot; + consumerGroupId + &quot; topic: &quot; + topic + &quot; lagDeltaSum: &quot; + topicOffsetList.sum)
            })
  
        })
    }
  </code></pre>