@@ -80,10 +80,13 @@ type KafkaSpec struct {
 	// +kubebuilder:pruning:PreserveUnknownFields
 	Template corev1.PodTemplateSpec `json:"template"`
 
-	// ArtifactURITemplate is rendered against the same context on a successful
-	// backup and recorded on the produced Backup's status.artifact.uri, so the
-	// stored object's location is introspectable. Ignored on restore; empty
-	// leaves status.artifact unset.
+	// ArtifactURITemplate is rendered against the same context, injected into the
+	// Job as ARTIFACT_URI (the single source of truth for the object location),
+	// and recorded on the produced Backup's status.artifact.uri so restore and
+	// cleanup act on the exact object the backup wrote. The driver requires it:
+	// reconcileKafka fails the BackupJob when it is empty, since nothing would
+	// record where the export is stored. The shipped cozy-default-kafka strategy
+	// always sets it; a custom Kafka strategy that stores backups must too.
 	// +optional
 	ArtifactURITemplate string `json:"artifactURITemplate,omitempty"`
 }
 
@@ -409,6 +409,18 @@ func (r *BackupJobReconciler) createKafkaBackupArtifact(
 		if getErr := r.Get(ctx, types.NamespacedName{Namespace: backup.Namespace, Name: backup.Name}, existing); getErr != nil {
 			return nil, getErr
 		}
+		// Adopt an existing Backup only when it describes the same application.
+		// A retained Backup whose BackupJob was reaped, then reused for a
+		// different application of the same name, would otherwise be reported as
+		// this run's success while describing something else (mirrors the
+		// RabbitMQ driver's guard).
+		if existing.Spec.ApplicationRef.Kind != backup.Spec.ApplicationRef.Kind ||
+			existing.Spec.ApplicationRef.Name != backup.Spec.ApplicationRef.Name {
+			return nil, fmt.Errorf("Backup %s/%s already exists for a different application (%s/%s, not %s/%s)",
+				backup.Namespace, backup.Name,
+				existing.Spec.ApplicationRef.Kind, existing.Spec.ApplicationRef.Name,
+				backup.Spec.ApplicationRef.Kind, backup.Spec.ApplicationRef.Name)
+		}
 		return existing, nil
 	}
 	return backup, nil
 
@@ -307,6 +307,37 @@ func TestCreateKafkaBackupArtifact_StampsScope(t *testing.T) {
 	}
 }
 
+// TestCreateKafkaBackupArtifact_RejectsDifferentApp pins the adoption guard: a
+// same-named Backup left over for a different application is not reported as this
+// run's success (mirrors the RabbitMQ driver).
+func TestCreateKafkaBackupArtifact_RejectsDifferentApp(t *testing.T) {
+	testScheme := runtime.NewScheme()
+	_ = scheme.AddToScheme(testScheme)
+	_ = backupsv1alpha1.AddToScheme(testScheme)
+
+	// A retained Backup named "bj" that describes application "other".
+	stale := &backupsv1alpha1.Backup{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj"},
+		Spec:       backupsv1alpha1.BackupSpec{ApplicationRef: kafkaAppRef("other")},
+	}
+	c := clientfake.NewClientBuilder().WithScheme(testScheme).
+		WithStatusSubresource(&backupsv1alpha1.Backup{}).WithObjects(stale).Build()
+	r := &BackupJobReconciler{Client: c, Scheme: testScheme, Recorder: record.NewFakeRecorder(10)}
+
+	bj := &backupsv1alpha1.BackupJob{
+		ObjectMeta: metav1.ObjectMeta{Namespace: "tenant", Name: "bj"},
+		Spec:       backupsv1alpha1.BackupJobSpec{ApplicationRef: kafkaAppRef("src")},
+	}
+	resolved := &ResolvedBackupConfig{
+		StrategyRef: corev1.TypedLocalObjectReference{Kind: strategyv1alpha1.KafkaStrategyKind, Name: "cozy-default-kafka"},
+	}
+	if _, err := r.createKafkaBackupArtifact(context.Background(), bj, resolved, "s3://bkt/tenant/src/bj/kafka-metadata.txt"); err == nil {
+		t.Fatal("expected an error adopting a same-named Backup for a different application, got nil")
+	} else if !strings.Contains(err.Error(), "different application") {
+		t.Fatalf("error = %v, want it to mention a different application", err)
+	}
+}
+
 // kafkaBackup builds a Backup fixture that has already recorded its metadata
 // object, for the cleanup path (whose delete is keyed on status.artifact.uri).
 func kafkaBackup(name, namespace string) *backupsv1alpha1.Backup {