feast-dev
diff --git a/‎ingestion/src/main/java/feast/ingestion/ImportJob.java‎
Lines changed: 2 additions & 0 deletions b/‎ingestion/src/main/java/feast/ingestion/ImportJob.java‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎ingestion/src/main/java/feast/ingestion/transform/ReadFromSource.java‎
Lines changed: 16 additions & 1 deletion b/‎ingestion/src/main/java/feast/ingestion/transform/ReadFromSource.java‎
Lines changed: 16 additions & 1 deletion
@@ -120,6 +120,8 @@ public static PipelineResult runPipeline(ImportOptions options) throws IOExcepti
               "ReadFeatureRowFromSource",
               ReadFromSource.newBuilder()
                   .setSource(source)
+                  .setConsumerGroupId(
+                      ReadFromSource.generateConsumerGroupId(source.getType(), store.getName()))
                   .setSuccessTag(FEATURE_ROW_OUT)
                   .setFailureTag(DEADLETTER_OUT)
                   .build());
 
@@ -23,6 +23,7 @@
 import feast.ingestion.transform.fn.KafkaRecordToFeatureRowDoFn;
 import feast.storage.api.writer.FailedElement;
 import feast.types.FeatureRowProto.FeatureRow;
+import java.util.Optional;
 import org.apache.beam.sdk.io.kafka.KafkaIO;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.ParDo;
@@ -41,6 +42,8 @@ public abstract class ReadFromSource extends PTransform<PBegin, PCollectionTuple
 
   public abstract TupleTag<FailedElement> getFailureTag();
 
+  public abstract Optional<String> getConsumerGroupId();
+
   public static Builder newBuilder() {
     return new AutoValue_ReadFromSource.Builder();
   }
@@ -54,6 +57,8 @@ public abstract static class Builder {
 
     public abstract Builder setFailureTag(TupleTag<FailedElement> failureTag);
 
+    public abstract Builder setConsumerGroupId(String consumerGroupId);
+
     abstract ReadFromSource autobuild();
 
     public ReadFromSource build() {
@@ -83,7 +88,10 @@ public PCollectionTuple expand(PBegin input) {
                 .withConsumerConfigUpdates(
                     ImmutableMap.of(
                         "group.id",
-                        generateConsumerGroupId(input.getPipeline().getOptions().getJobName())))
+                        getConsumerGroupId().isPresent()
+                            ? getConsumerGroupId().get()
+                            : generateConsumerGroupId(
+                                input.getPipeline().getOptions().getJobName())))
                 .withReadCommitted()
                 .commitOffsetsInFinalize())
         .apply(
@@ -99,4 +107,11 @@ public PCollectionTuple expand(PBegin input) {
   private String generateConsumerGroupId(String jobName) {
     return "feast_import_job_" + jobName;
   }
+
+  /** Get a stable Kafka consumer group ID, so restarting jobs can resume offsets. */
+  public static String generateConsumerGroupId(SourceType sourceType, String storeName) {
+    return String.format("feast_import_job_%s_%s", sourceType, storeName)
+        .replaceAll(" ", "_")
+        .toLowerCase();
+  }
 }