diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java
index f07e9aee1..aa8fe95bb 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -61,6 +61,7 @@ import org.springframework.util.ObjectUtils;
  * @author Jon Schneider
  * @author Thomas Cheyney
  * @author Gary Russell
+ * @author Lars Bilger
  */
 public class KafkaBinderMetrics
 		implements MeterBinder, ApplicationListener<BindingCreatedEvent> {
@@ -209,14 +210,19 @@ public class KafkaBinderMetrics
 				.endOffsets(topicPartitions);
 
 		final Map<TopicPartition, OffsetAndMetadata> committedOffsets = metadataConsumer.committed(endOffsets.keySet());
+		final Map<TopicPartition, Long> beginningOffsets = metadataConsumer.beginningOffsets(endOffsets.keySet());
 
 		for (Map.Entry<TopicPartition, Long> endOffset : endOffsets
 				.entrySet()) {
 			OffsetAndMetadata current = committedOffsets.get(endOffset.getKey());
+			Long beginningOffset = beginningOffsets.get(endOffset.getKey());
 			lag += endOffset.getValue();
 			if (current != null) {
 				lag -= current.offset();
 			}
+			else if (beginningOffset != null) {
+				lag -= beginningOffset;
+			}
 		}
 		return lag;
 	}
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java
index a1e3a432c..af82a0b58 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java
@@ -17,6 +17,7 @@
 package org.springframework.cloud.stream.binder.kafka;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -49,6 +50,7 @@ import static org.mockito.Mockito.mock;
  * @author Henryk Konsek
  * @author Thomas Cheyney
  * @author Soby Chacko
+ * @author Lars Bilger
  */
 public class KafkaBinderMetricsTest {
 
@@ -254,6 +256,29 @@ public class KafkaBinderMetricsTest {
 				ArgumentMatchers.eq("group2-metrics"), ArgumentMatchers.any());
 	}
 
+	@Test
+	public void usesBeginningOffsetIfNoCommittedOffsetFound() {
+		org.mockito.BDDMockito
+			.given(consumer.committed(ArgumentMatchers.anySet()))
+			.willReturn(Collections.emptyMap());
+		final Map<TopicPartition, Long> beginnings = new HashMap<>();
+		TopicPartition topicPartition = new TopicPartition(TEST_TOPIC, 0);
+		beginnings.put(topicPartition, 500L);
+		org.mockito.BDDMockito
+			.given(consumer.beginningOffsets(ArgumentMatchers.anySet()))
+			.willReturn(beginnings);
+		List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
+		topicsInUse.put(TEST_TOPIC,
+						new TopicInformation("group1-metrics", partitions, false));
+		org.mockito.BDDMockito.given(consumer.partitionsFor(TEST_TOPIC))
+							.willReturn(partitions);
+		metrics.bindTo(meterRegistry);
+		assertThat(meterRegistry.getMeters()).hasSize(1);
+		assertThat(meterRegistry.get(KafkaBinderMetrics.OFFSET_LAG_METRIC_NAME)
+								.tag("group", "group1-metrics").tag("topic", TEST_TOPIC).gauge().value())
+			.isEqualTo(500.0);
+	}
+
 	private List<PartitionInfo> partitions(Node... nodes) {
 		List<PartitionInfo> partitions = new ArrayList<>();
 		for (int i = 0; i < nodes.length; i++) {