fix(rdf): ensure Fuseki dataset before indexing and report failures correctly

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/rdf/RdfIndexApp.java

@@ -142,6 +142,19 @@ public void execute(JobExecutionContext jobExecutionContext) {
       return;
     }
 
+    try {
+      rdfRepository.ensureStorageReady();
+    } catch (Exception e) {
+      LOG.error("RDF storage is not ready; aborting indexing job", e);
+      updateJobStatus(EventPublisherJob.Status.FAILED);
+      jobData.setFailure(
+          new IndexingError()
+              .withErrorSource(IndexingError.ErrorSource.JOB)
+              .withMessage("RDF storage is not ready: " + e.getMessage()));
+      sendUpdates(jobExecutionContext, true);
+      return;
+    }
+
     String jobName = jobExecutionContext.getJobDetail().getKey().getName();
     if (jobName.equals(ON_DEMAND_JOB)) {
       Map<String, Object> jsonAppConfig = JsonUtils.convertValue(jobData, Map.class);

openmetadata-service/src/main/java/org/openmetadata/service/apps/bundles/rdf/distributed/DistributedRdfIndexExecutor.java

@@ -128,6 +128,7 @@ public void execute(EventPublisherJob jobConfiguration) throws InterruptedExcept
 
   public void joinJob(RdfIndexJob job, EventPublisherJob jobConfiguration)
       throws InterruptedException {
+    RdfRepository.getInstance().ensureStorageReady();
     currentJob = job;
     coordinatorOwnedJob = false;
     stopped.set(false);

openmetadata-service/src/main/java/org/openmetadata/service/rdf/RdfRepository.java

@@ -106,6 +106,13 @@ public String getBaseUri() {
     return config.getBaseUri().toString();
   }
 
+  public void ensureStorageReady() {
+    if (!isEnabled()) {
+      return;
+    }
+    storageService.ensureStorageReady();
+  }
+
   public void createOrUpdate(EntityInterface entity) {
     if (!isEnabled()) {
       return;
@@ -147,6 +154,9 @@ public void createOrUpdate(EntityInterface entity) {
           entity.getEntityReference().getType(),
           entity.getFullyQualifiedName(),
           e);
+      // Rethrow so callers (e.g. RdfBatchProcessor) can count this as a failure instead of
+      // reporting a false success. Entity-hook callers (RdfUpdater) already wrap in try/catch.
+      throw new RuntimeException("Failed to create/update entity in RDF", e);
     }
   }
 

openmetadata-service/src/main/java/org/openmetadata/service/rdf/storage/JenaFusekiStorage.java

@@ -39,20 +39,26 @@ public class JenaFusekiStorage implements RdfStorageInterface {
 
   private final RDFConnection connection;
   private final String baseUri;
+  private final String endpoint;
+  private final String username;
+  private final String password;
 
   public JenaFusekiStorage(RdfConfiguration config) {
     this.baseUri =
         config.getBaseUri() != null ? config.getBaseUri().toString() : "https://open-metadata.org/";
 
-    String endpoint =
+    this.endpoint =
         config.getRemoteEndpoint() != null && !config.getRemoteEndpoint().toString().isEmpty()
             ? config.getRemoteEndpoint().toString()
             : "http://openmetadata-fuseki:3030/openmetadata";
+    this.username = config.getUsername();
+    this.password = config.getPassword();
 
-    // Ensure the dataset exists before connecting
-    ensureDatasetExists(endpoint, config.getUsername(), config.getPassword());
+    // Best-effort attempt to create the dataset at startup; callers should invoke
+    // ensureStorageReady() before running work to recover from later restarts of the RDF server.
+    ensureDatasetExists(endpoint, username, password);
 
-    if (config.getUsername() != null && config.getPassword() != null) {
+    if (username != null && password != null) {
       java.net.http.HttpClient httpClient =
           java.net.http.HttpClient.newBuilder()
               .authenticator(
@@ -74,6 +80,30 @@ protected java.net.PasswordAuthentication getPasswordAuthentication() {
     loadOntology();
   }
 
+  @Override
+  public void ensureStorageReady() {
+    if (testConnection()) {
+      LOG.debug("Fuseki dataset at {} is accessible", endpoint);
+      return;
+    }
+
+    LOG.warn(
+        "Fuseki dataset at {} is not accessible; attempting to (re)create it before running",
+        endpoint);
+    ensureDatasetExists(endpoint, username, password);
+
+    if (!testConnection()) {
+      throw new IllegalStateException(
+          String.format(
+              "RDF storage is not accessible at %s after attempting dataset creation. "
+                  + "Verify the configured RDF endpoint URL, credentials, that the Fuseki dataset "
+                  + "exists, and that the configured user has permission to create it.",
+              endpoint));
+    }
+    LOG.info("Fuseki dataset at {} is now ready", endpoint);
+    loadOntology();
+  }
+
   /**
    * Ensures the Fuseki dataset exists, creating it if necessary.
    * Parses the endpoint URL to extract the server base URL and dataset name,

openmetadata-service/src/main/java/org/openmetadata/service/rdf/storage/RdfStorageInterface.java

@@ -74,6 +74,14 @@ void storeRelationship(
    */
   boolean testConnection();
 
+  /**
+   * Verify the underlying storage is reachable and the configured dataset/graph is accessible,
+   * attempting to create it if missing. Implementations must throw if the storage cannot be
+   * brought to a ready state so callers can surface a clear error instead of silently producing
+   * partial results.
+   */
+  default void ensureStorageReady() {}
+
   /**
    * Get storage type identifier
    */