diff --git a/.github/workflows/maven.yml b/.github/workflows/maven.yml
index dc3c906c433..9af159c75ea 100644
--- a/.github/workflows/maven.yml
+++ b/.github/workflows/maven.yml
@@ -25,8 +25,7 @@ on:
       - beta
       - master
       - staging
-      - mlh-1477-staging
-      - mlh1638-staging
+      - mlh-1477-staging-dlq
 
 jobs:
   # Detect what changed to optimize workflow execution
diff --git a/LOCAL_SETUP.md b/LOCAL_SETUP.md
index 082866f8cce..07793a28fb6 100644
--- a/LOCAL_SETUP.md
+++ b/LOCAL_SETUP.md
@@ -9,6 +9,8 @@ This guide will help you set up Atlas for local development.
 - Maven 3.8+ 
 - Docker (via Colima for macOS)
 - Git
+- Get the source code from the AtlanHQ repository (An override of Apache Atlas)
+- Download the zip and configuration artifacts from https://atlanhq.atlassian.net/wiki/spaces/c873aeb606dd4834a95d9909a757bfa6/pages/800424446/How+to+run+Atlas+on+the+local+machine
 
 ### Java Setup
 1. Install Java 17:
@@ -148,3 +150,4 @@ For more detailed information, refer to:
 - The build command skips tests and various checks for faster development builds
 - For production builds, remove the skip flags
 - Keep your GitHub PAT token secure and never commit it to version control
+- Adjust memory and CPU settings in Colima based on your machine's capabilities
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 80cdd938bbc..6fc6ef797d8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -720,7 +720,7 @@
         <httpcomponents-httpcore.version>4.4.13</httpcomponents-httpcore.version>
         <jackson.databind.version>2.13.4.2</jackson.databind.version>
         <jackson.version>2.12.4</jackson.version>
-        <janusgraph.version>1.0.2-atlan-SNAPSHOT</janusgraph.version>
+        <janusgraph.version>1.0.2-atlan</janusgraph.version>
         <janusgraph.cassandra.version>0.5.3</janusgraph.cassandra.version>
         <jaxb.api.version>2.3.1</jaxb.api.version>
         <javax-inject.version>1</javax-inject.version>
diff --git a/webapp/pom.xml b/webapp/pom.xml
index 71b6ddaed5e..613c01fc4d5 100755
--- a/webapp/pom.xml
+++ b/webapp/pom.xml
@@ -375,9 +375,19 @@
             <artifactId>jersey-multipart</artifactId>
         </dependency>
 
+        <!-- Mockito upgraded to support Java 17 -->
         <dependency>
             <groupId>org.mockito</groupId>
-            <artifactId>mockito-all</artifactId>
+            <artifactId>mockito-core</artifactId>
+            <version>4.11.0</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <version>4.11.0</version>
+            <scope>test</scope>
         </dependency>
 
         <!--<dependency>
diff --git a/webapp/src/main/java/org/apache/atlas/web/rest/DLQAdminController.java b/webapp/src/main/java/org/apache/atlas/web/rest/DLQAdminController.java
new file mode 100644
index 00000000000..ae1cd1c4f05
--- /dev/null
+++ b/webapp/src/main/java/org/apache/atlas/web/rest/DLQAdminController.java
@@ -0,0 +1,28 @@
+package org.apache.atlas.web.rest;
+
+import org.apache.atlas.web.service.DLQReplayService;
+import org.apache.atlas.web.util.Servlets;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import javax.inject.Singleton;
+import javax.ws.rs.*;
+import javax.ws.rs.core.MediaType;
+import java.util.Map;
+
+@Path("dlq")
+@Singleton
+@Service
+@Consumes({Servlets.JSON_MEDIA_TYPE, MediaType.APPLICATION_JSON})
+@Produces({Servlets.JSON_MEDIA_TYPE, MediaType.APPLICATION_JSON})
+public class DLQAdminController {
+
+    @Autowired
+    private DLQReplayService dlqReplayService;
+
+    @GET
+    @Path("/replay/status")
+    public Map<String, Object> getStatus() {
+        return dlqReplayService.getStatus();
+    }
+}
diff --git a/webapp/src/main/java/org/apache/atlas/web/service/DLQReplayService.java b/webapp/src/main/java/org/apache/atlas/web/service/DLQReplayService.java
new file mode 100644
index 00000000000..01f717fc8eb
--- /dev/null
+++ b/webapp/src/main/java/org/apache/atlas/web/service/DLQReplayService.java
@@ -0,0 +1,779 @@
+package org.apache.atlas.web.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.core.JsonParser;
+import java.io.IOException;
+import org.apache.atlas.ApplicationProperties;
+import org.apache.atlas.AtlasException;
+import org.apache.atlas.repository.graphdb.janus.AtlasJanusGraph;
+import org.apache.kafka.clients.consumer.*;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.janusgraph.diskstorage.Backend;
+import org.janusgraph.diskstorage.BaseTransaction;
+import org.janusgraph.diskstorage.StandardIndexProvider;
+import org.janusgraph.diskstorage.TemporaryBackendException;
+import org.janusgraph.diskstorage.configuration.Configuration;
+import org.janusgraph.diskstorage.dlq.DLQEntry;
+import org.janusgraph.diskstorage.dlq.SerializableIndexMutation;
+import org.janusgraph.diskstorage.keycolumnvalue.StoreFeatures;
+import org.janusgraph.diskstorage.util.StandardBaseTransactionConfig;
+import org.janusgraph.diskstorage.indexing.IndexProvider;
+import org.janusgraph.diskstorage.indexing.IndexEntry;
+import org.janusgraph.diskstorage.indexing.IndexMutation;
+import org.janusgraph.diskstorage.indexing.KeyInformation;
+import org.janusgraph.diskstorage.es.ElasticSearchIndex;
+import org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration;
+import org.janusgraph.graphdb.database.IndexSerializer;
+import org.janusgraph.graphdb.database.StandardJanusGraph;
+import org.janusgraph.graphdb.database.index.IndexInfoRetriever;
+import org.janusgraph.graphdb.transaction.StandardJanusGraphTx;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Service for replaying DLQ messages back to Elasticsearch.
+ * Uses Kafka consumer groups to ensure only one instance processes messages.
+ */
+@Service
+public class DLQReplayService {
+
+    private static final Logger log = LoggerFactory.getLogger(DLQReplayService.class);
+
+    private String bootstrapServers;
+    @Value("${atlas.kafka.dlq.topic:ATLAS_ES_DLQ}")
+    private String dlqTopic="ATLAS_ES_DLQ";
+
+    @Value("${atlas.kafka.dlq.consumerGroupId:atlas_dq_replay_group}")
+    private String consumerGroupId= "atlas_dq_replay_group";
+
+    @Value("${atlas.kafka.dlq.maxRetries:3}")
+    private int maxRetries = 3;
+
+    // Kafka consumer configuration
+    @Value("${atlas.kafka.dlq.maxPollRecords:10}")
+    private int maxPollRecords = 10;
+
+    @Value("${atlas.kafka.dlq.maxPollIntervalMs:600000}")
+    private int maxPollIntervalMs = 600000; // 10 minutes
+
+    @Value("${atlas.kafka.dlq.sessionTimeoutMs:90000}")
+    private int sessionTimeoutMs = 90000; // 90 seconds
+
+    @Value("${atlas.kafka.dlq.heartbeatIntervalMs:30000}")
+    private int heartbeatIntervalMs = 30000; // 30 seconds
+
+    // Timing configuration
+    @Value("${atlas.kafka.dlq.pollTimeoutSeconds:5}")
+    private int pollTimeoutSeconds = 5;
+
+    @Value("${atlas.kafka.dlq.shutdownWaitSeconds:60}")
+    private int shutdownWaitSeconds = 60;
+
+    @Value("${atlas.kafka.dlq.consumerCloseTimeoutSeconds:30}")
+    private int consumerCloseTimeoutSeconds = 30;
+
+    @Value("${atlas.kafka.dlq.errorBackoffMs:10000}")
+    private int errorBackoffMs = 10000; // 10 seconds (for permanent exceptions)
+
+    // Exponential backoff configuration for temporary exceptions
+    @Value("${atlas.kafka.dlq.exponentialBackoff.baseDelayMs:1000}")
+    private int exponentialBackoffBaseDelayMs = 1000; // 1 second
+
+    @Value("${atlas.kafka.dlq.exponentialBackoff.maxDelayMs:60000}")
+    private int exponentialBackoffMaxDelayMs = 60000; // 60 seconds
+
+    @Value("${atlas.kafka.dlq.exponentialBackoff.multiplier:2.0}")
+    private double exponentialBackoffMultiplier = 2.0;
+
+    @Value("${atlas.kafka.dlq.trackerCleanupIntervalMs:300000}")
+    private long trackerCleanupIntervalMs = 300000; // 5 minutes
+
+    @Value("${atlas.kafka.dlq.trackerMaxAgeMs:3600000}")
+    private long trackerMaxAgeMs = 3600000; // 1 hour
+
+    @Value("${atlas.kafka.dlq.retryDelayMs:5000}")
+    private int retryDelayMs = 5000; // 5 seconds
+
+    private final ElasticSearchIndex esIndex;
+    protected final IndexSerializer indexSerializer;
+    private final ObjectMapper mapper;
+    private GraphDatabaseConfiguration graphConfig;
+    private StandardJanusGraph standardJanusGraph;
+
+    //Track retry attempts with timestamps for cleanup
+    private static class RetryTrackerEntry {
+        int retryCount;
+        long lastAttemptTime;
+
+        RetryTrackerEntry(int retryCount) {
+            this.retryCount = retryCount;
+            this.lastAttemptTime = System.currentTimeMillis();
+        }
+    }
+
+    private final Map<String, RetryTrackerEntry> retryTracker = new ConcurrentHashMap<>();
+
+    // Track exponential backoff delay per partition-offset for temporary exceptions (in-memory)
+    private final Map<String, Long> backoffTracker = new ConcurrentHashMap<>();
+
+    // P1 Fix: Track last cleanup time
+    private volatile long lastTrackerCleanupTime = System.currentTimeMillis();
+
+    private volatile KafkaConsumer<String, String> consumer;
+    private final AtomicBoolean isRunning = new AtomicBoolean(false);
+    private final AtomicBoolean isHealthy = new AtomicBoolean(true);
+    private volatile Thread replayThread;
+    private final AtomicInteger processedCount = new AtomicInteger(0);
+    private final AtomicInteger errorCount = new AtomicInteger(0);
+    private final AtomicInteger skippedCount = new AtomicInteger(0);
+
+    private final static String INDEX_NAME = "search";
+
+    public DLQReplayService(AtlasJanusGraph graph) throws AtlasException {
+        this.mapper = configureMapper();
+        // Extract ES configuration from existing graph
+        this.graphConfig = ((StandardJanusGraph)graph.getGraph()).getConfiguration();
+        this.standardJanusGraph = ((StandardJanusGraph)graph.getGraph());
+        this.bootstrapServers = ApplicationProperties.get().getString("atlas.graph.kafka.bootstrap.servers");
+        Configuration fullConfig = graphConfig.getConfiguration();
+        IndexProvider indexProvider = Backend.getImplementationClass(fullConfig.restrictTo(INDEX_NAME), fullConfig.get(GraphDatabaseConfiguration.INDEX_BACKEND,INDEX_NAME),
+                StandardIndexProvider.getAllProviderClasses());
+        StoreFeatures storeFeatures = graphConfig.getBackend().getStoreFeatures();
+        this.indexSerializer = new IndexSerializer(fullConfig, graphConfig.getSerializer(),
+                graphConfig.getBackend().getIndexInformation(), storeFeatures.isDistributed() && storeFeatures.isKeyOrdered());
+        esIndex = (ElasticSearchIndex) indexProvider;
+    }
+
+    /**
+     * Start replaying DLQ messages
+     */
+    @PostConstruct
+    public synchronized void startReplay() {
+        if (isRunning.get()) {
+            log.warn("DLQ replay is already running");
+            return;
+        }
+
+        log.info("Starting DLQ replay service for topic: {} with consumer group: {}", dlqTopic, consumerGroupId);
+
+        Properties consumerProps = new Properties();
+        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
+        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // Manual commit after success
+        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+        // Optimized settings for long-running message processing with pause/resume pattern
+        consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, String.valueOf(maxPollRecords));
+        consumerProps.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, String.valueOf(maxPollIntervalMs));
+        consumerProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, String.valueOf(sessionTimeoutMs));
+        consumerProps.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, String.valueOf(heartbeatIntervalMs));
+
+        consumer = new KafkaConsumer<>(consumerProps);
+        consumer.subscribe(Collections.singletonList(dlqTopic), new ConsumerRebalanceListener() {
+            @Override
+            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
+                log.warn("Consumer group partitions revoked. Partitions: {}", partitions);
+
+                // Clean up retry tracker and backoff tracker for revoked partitions to prevent memory leak
+                for (TopicPartition partition : partitions) {
+                    String partitionPrefix = partition.partition() + "-";
+                    retryTracker.keySet().removeIf(key -> key.startsWith(partitionPrefix));
+                    backoffTracker.keySet().removeIf(key -> key.startsWith(partitionPrefix));
+                    log.info("Cleaned up retry and backoff trackers for revoked partition: {}", partition);
+                }
+            }
+
+            @Override
+            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
+                log.info("Consumer group partitions assigned: {}", partitions);
+
+                // Log offset information for each partition
+                for (TopicPartition partition : partitions) {
+                    try {
+                        long endOffset = consumer.endOffsets(Collections.singleton(partition)).get(partition);
+                        long committedOffset = -1;
+                        OffsetAndMetadata committed = consumer.committed(Collections.singleton(partition)).get(partition);
+                        if (committed != null) {
+                            committedOffset = committed.offset();
+                        }
+                        long position = consumer.position(partition);
+
+                        log.info("Partition {} - End offset: {}, Committed offset: {}, Current position: {}, " +
+                                        "Messages available: {}",
+                                partition, endOffset, committedOffset, position,
+                                endOffset - position);
+                    } catch (Exception e) {
+                        log.error("Error checking offsets for partition: " + partition, e);
+                    }
+                }
+            }
+        });
+
+        isRunning.set(true);
+        isHealthy.set(true);
+
+        // Start processing in a separate thread
+        replayThread = new Thread(this::processMessages, "DLQ-Replay-Thread");
+        replayThread.setDaemon(false);
+        replayThread.start();
+
+        log.info("DLQ replay service started successfully");
+    }
+
+    /**
+     * Gracefully shutdown the DLQ replay service with proper thread join
+     */
+    @PreDestroy
+    public synchronized void shutdown() {
+        if (!isRunning.get()) {
+            log.info("DLQ replay service is not running, nothing to shutdown");
+            return;
+        }
+
+        log.info("Shutting down DLQ replay service...");
+        isRunning.set(false);
+
+        // Step 1: Interrupt the consumer's poll() to wake it up
+        if (consumer != null) {
+            try {
+                consumer.wakeup();
+                log.info("Sent wakeup signal to Kafka consumer");
+            } catch (Exception e) {
+                log.error("Error sending wakeup to consumer", e);
+            }
+        }
+
+        // Step 2: Wait for the replay thread to finish current work
+        if (replayThread != null && replayThread.isAlive()) {
+            try {
+                // P0 Fix: Calculate appropriate timeout (consumer close timeout + buffer for processing)
+                long shutdownTimeoutMs = TimeUnit.SECONDS.toMillis(shutdownWaitSeconds);
+                log.info("Waiting up to {}ms for replay thread to complete current work...", shutdownTimeoutMs);
+
+                replayThread.join(shutdownTimeoutMs);
+
+                if (replayThread.isAlive()) {
+                    log.error("Replay thread did not terminate within timeout of {}ms! " +
+                                    "Current message processing may be interrupted. Thread state: {}",
+                            shutdownTimeoutMs, replayThread.getState());
+                    // Thread will be forcefully terminated when JVM exits
+                } else {
+                    log.info("Replay thread terminated gracefully");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.warn("Interrupted while waiting for replay thread to finish");
+            }
+        }
+
+        // Step 3: Close consumer (thread should be done by now)
+        if (consumer != null) {
+            try {
+                consumer.close(Duration.ofSeconds(consumerCloseTimeoutSeconds));
+                log.info("Kafka consumer closed successfully");
+            } catch (Exception e) {
+                log.error("Error closing Kafka consumer during shutdown", e);
+            }
+        }
+
+        log.info("DLQ replay service shutdown complete. Total processed: {}, Total errors: {}, Total skipped: {}",
+                processedCount.get(), errorCount.get(), skippedCount.get());
+    }
+
+    /**
+     * Process messages from the DLQ topic using pause/resume pattern
+     */
+    private void processMessages() {
+        log.info("DLQ replay thread started, polling for messages...");
+
+        try {
+            while (isRunning.get()) {
+                try {
+                    // P1 Fix: Periodic cleanup of stale tracker entries
+                    cleanupStaleTrackersIfNeeded();
+
+                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(pollTimeoutSeconds));
+
+                    if (records.isEmpty()) {
+                        // Log why we got no records
+                        for (TopicPartition partition : consumer.assignment()) {
+                            try {
+                                long currentPosition = consumer.position(partition);
+                                long endOffset = consumer.endOffsets(Collections.singleton(partition)).get(partition);
+                                if (currentPosition >= endOffset) {
+                                    log.debug("No messages available - Partition {} at end (Position: {}, End: {})",
+                                            partition, currentPosition, endOffset);
+                                } else {
+                                    log.info("No messages returned despite availability - Partition {} (Position: {}, End: {}, Available: {})",
+                                            partition, currentPosition, endOffset, endOffset - currentPosition);
+                                }
+                            } catch (Exception e) {
+                                log.error("Error checking position after empty poll", e);
+                            }
+                        }
+                        continue;
+                    }
+
+                    log.debug("Received {} DLQ messages to replay", records.count());
+
+                    // PAUSE consumption immediately to prevent timeout during processing
+                    Set<TopicPartition> pausedPartitions = consumer.assignment();
+                    consumer.pause(pausedPartitions);
+                    log.info("Paused consumption on partitions: {} to process messages", pausedPartitions);
+
+                    Long failedOffset = null;  // Track if we need to seek back
+                    TopicPartition failedPartition = null;
+
+                    try {
+                        // Now process without time pressure - heartbeats continue automatically
+                        for (ConsumerRecord<String, String> record : records) {
+                            String retryKey = record.partition() + "-" + record.offset();
+
+                            try {
+                                log.info("Processing DLQ entry at offset: {} from partition: {}",
+                                        record.offset(), record.partition());
+
+                                long processingStartTime = System.currentTimeMillis();
+                                replayDLQEntry(record.value());
+                                long processingTime = System.currentTimeMillis() - processingStartTime;
+
+                                processedCount.incrementAndGet();
+
+                                // Success - remove from retry tracker, reset backoff, and commit offset
+                                retryTracker.remove(retryKey);
+                                resetExponentialBackoff(retryKey);
+
+                                Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(
+                                        new TopicPartition(record.topic(), record.partition()),
+                                        new OffsetAndMetadata(record.offset() + 1)
+                                );
+                                consumer.commitSync(offsets);
+
+                                log.debug("Successfully replayed DLQ entry (offset: {}, partition: {}) in {}ms",
+                                        record.offset(), record.partition(), processingTime);
+
+                            } catch (TemporaryBackendException temporaryBackendException) {
+                                // Treat temporary backend exceptions as transient - will retry with exponential backoff
+                                errorCount.incrementAndGet();
+
+                                // Calculate exponential backoff delay
+                                long backoffDelay = calculateExponentialBackoff(retryKey);
+
+                                log.warn("Temporary backend exception while replaying DLQ entry (offset: {}, partition: {}). " +
+                                                "Will retry on next poll after {}ms backoff (exponential). STOPPING batch processing to prevent skipping this message. Error: {}",
+                                        record.offset(), record.partition(), backoffDelay, temporaryBackendException.getMessage());
+
+                                // Mark for seek-back - consumer position has already advanced past this offset
+                                failedOffset = record.offset();
+                                failedPartition = new TopicPartition(record.topic(), record.partition());
+
+                                Thread.sleep(backoffDelay);
+                                break;
+                            } catch (Exception e) {
+                                errorCount.incrementAndGet();
+
+                                //Track retry attempts with timestamp for this specific offset
+                                RetryTrackerEntry entry = retryTracker.get(retryKey);
+                                int retryCount;
+                                if (entry == null) {
+                                    retryCount = 1;
+                                    retryTracker.put(retryKey, new RetryTrackerEntry(retryCount));
+                                } else {
+                                    retryCount = entry.retryCount + 1;
+                                    entry.retryCount = retryCount;
+                                    entry.lastAttemptTime = System.currentTimeMillis();
+                                }
+
+                                if (retryCount >= maxRetries) {
+                                    //Clean up both trackers for poison pill
+                                    log.error("DLQ entry at offset {} partition {} failed {} times (max retries reached). " +
+                                                    "SKIPPING this message to prevent partition blockage. Error: {}",
+                                            record.offset(), record.partition(), retryCount, e.getMessage(), e);
+
+                                    skippedCount.incrementAndGet();
+                                    retryTracker.remove(retryKey);
+                                    resetExponentialBackoff(retryKey); // P0 Fix: Clean up backoff tracker too
+
+                                    // Commit offset to move past poison pill
+                                    Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(
+                                            new TopicPartition(record.topic(), record.partition()),
+                                            new OffsetAndMetadata(record.offset() + 1)
+                                    );
+                                    try {
+                                        consumer.commitSync(offsets);
+                                        log.warn("Committed offset {} to skip poison pill", record.offset() + 1);
+                                    } catch (Exception commitEx) {
+                                        log.error("Failed to commit offset after skipping poison pill", commitEx);
+                                    }
+                                    // Continue to next record after skipping poison pill
+                                } else {
+                                    //Add delay before retrying to avoid tight retry loop
+                                    log.warn("Failed to replay DLQ entry (offset: {}, partition: {}). Retry {}/{}. " +
+                                                    "STOPPING batch processing to prevent skipping this message. Will retry on next poll after {}ms delay. Error: {}",
+                                            record.offset(), record.partition(), retryCount, maxRetries, retryDelayMs, e.getMessage());
+
+                                    // Mark for seek-back - consumer position has already advanced past this offset
+                                    failedOffset = record.offset();
+                                    failedPartition = new TopicPartition(record.topic(), record.partition());
+
+                                    //Sleep before breaking to avoid tight retry loop
+                                    Thread.sleep(retryDelayMs);
+                                    break;
+                                }
+                            }
+                        }
+                    } finally {
+                        // The consumer's position advances when records are polled, not when committed
+                        // We need to rewind to retry the failed message
+                        if (failedOffset != null && failedPartition != null) {
+                            try {
+                                consumer.seek(failedPartition, failedOffset);
+                                log.info("Seeked back to offset {} on partition {} to retry failed message",
+                                        failedOffset, failedPartition);
+                            } catch (Exception seekEx) {
+                                log.error("Failed to seek back to offset {} on partition {}. Message may be skipped!",
+                                        failedOffset, failedPartition, seekEx);
+                            }
+                        }
+
+                        // RESUME consumption - always do this even if processing failed
+                        consumer.resume(pausedPartitions);
+                        log.info("Resumed consumption on partitions: {}", pausedPartitions);
+                    }
+
+                } catch (WakeupException e) {
+                    // Expected during shutdown - exit gracefully
+                    log.info("Kafka consumer wakeup called, exiting processing loop");
+                    break;
+                } catch (Exception e) {
+                    log.error("Error in DLQ replay processing loop", e);
+                    try {
+                        // Back off before retry
+                        Thread.sleep(errorBackoffMs);
+                    } catch (InterruptedException ie) {
+                        Thread.currentThread().interrupt();
+                        log.warn("DLQ replay thread interrupted during error recovery");
+                        break;
+                    }
+                }
+            }
+        } catch (Exception e) {
+            log.error("Fatal error in DLQ replay thread - marking service as unhealthy", e);
+            isHealthy.set(false);
+        } finally {
+            boolean wasRunning = isRunning.get();
+            isRunning.set(false);
+
+            if (wasRunning && !isHealthy.get()) {
+                log.error("DLQ replay thread terminated unexpectedly! Service is unhealthy. Pod should be restarted.");
+            }
+
+            log.info("DLQ replay thread finished. Total processed: {}, Total errors: {}, Total skipped: {}",
+                    processedCount.get(), errorCount.get(), skippedCount.get());
+        }
+    }
+
+    /**
+     * P1 Fix: Periodic cleanup of stale tracker entries to prevent memory leak
+     */
+    private void cleanupStaleTrackersIfNeeded() {
+        long currentTime = System.currentTimeMillis();
+        if (currentTime - lastTrackerCleanupTime > trackerCleanupIntervalMs) {
+            int retryTrackerSizeBefore = retryTracker.size();
+            int backoffTrackerSizeBefore = backoffTracker.size();
+
+            long cutoffTime = currentTime - trackerMaxAgeMs;
+
+            // Clean up stale retry tracker entries
+            retryTracker.entrySet().removeIf(entry -> entry.getValue().lastAttemptTime < cutoffTime);
+
+            // Clean up stale backoff tracker entries (we don't have timestamps, so this is best-effort)
+            // In practice, backoff entries get cleaned on success or poison pill, so this is just safety net
+            if (backoffTracker.size() > retryTracker.size() * 2) {
+                log.warn("Backoff tracker size ({}) is much larger than retry tracker size ({}), clearing orphaned entries",
+                        backoffTracker.size(), retryTracker.size());
+                // Keep only entries that have corresponding retry tracker entries
+                Set<String> activeRetryKeys = retryTracker.keySet();
+                backoffTracker.keySet().removeIf(key -> !activeRetryKeys.contains(key));
+            }
+
+            int retryTrackerCleaned = retryTrackerSizeBefore - retryTracker.size();
+            int backoffTrackerCleaned = backoffTrackerSizeBefore - backoffTracker.size();
+
+            if (retryTrackerCleaned > 0 || backoffTrackerCleaned > 0) {
+                log.info("Cleaned up stale tracker entries: retry tracker {} -> {} (removed {}), backoff tracker {} -> {} (removed {})",
+                        retryTrackerSizeBefore, retryTracker.size(), retryTrackerCleaned,
+                        backoffTrackerSizeBefore, backoffTracker.size(), backoffTrackerCleaned);
+            }
+
+            lastTrackerCleanupTime = currentTime;
+        }
+    }
+
+    /**
+     * Replay a single DLQ entry
+     */
+    private void replayDLQEntry(String dlqJson) throws Exception {
+        long startTime = System.currentTimeMillis();
+        StandardJanusGraphTx standardJanusGraphTx = null;
+        BaseTransaction replayTx = null;
+
+        try {
+            DLQEntry entry = mapper.readValue(dlqJson, DLQEntry.class);
+            log.info("Replaying DLQ entry for index: {}, store: {}", entry.getIndexName(), entry.getStoreName());
+
+            // Create key information retriever
+            log.info("Creating key information retriever for index: {}", entry.getIndexName());
+            standardJanusGraphTx = (StandardJanusGraphTx) this.standardJanusGraph.newTransaction();
+            IndexInfoRetriever keyInfo = this.indexSerializer.getIndexInfoRetriever(standardJanusGraphTx);
+
+            // Reconstruct mutations from serialized form
+            log.info("Starting mutation reconstruction for index: {}", entry.getIndexName());
+            Map<String, Map<String, IndexMutation>> mutations = reconstructMutations(entry, keyInfo.get(INDEX_NAME));
+            log.info("Completed mutation reconstruction in {}ms", System.currentTimeMillis() - startTime);
+
+
+            log.info("JanusGraph transaction committed (metadata lookup)");
+
+            // Create a new transaction for replay
+            log.info("Beginning ES transaction for index: {}", entry.getIndexName());
+            replayTx = esIndex.beginTransaction(
+                    new StandardBaseTransactionConfig.Builder().commitTime(Instant.now()).build()
+            );
+
+            // This is the same method that originally failed - now we're replaying it!
+            log.info("Starting ES mutation for index: {}", entry.getIndexName());
+            long mutateStartTime = System.currentTimeMillis();
+
+            esIndex.mutate(mutations, keyInfo.get(INDEX_NAME), replayTx);
+
+            log.info("ES mutation completed in {}ms, committing ES transaction", System.currentTimeMillis() - mutateStartTime);
+            long commitStartTime = System.currentTimeMillis();
+            // Commit JanusGraph transaction BEFORE ES mutation to ensure read-only metadata lookup is complete
+            // This way if ES mutation fails, we only rollback ES, not the metadata lookup
+            standardJanusGraphTx.commit();
+            standardJanusGraphTx = null; // Mark as committed so we don't rollback in catch block
+            replayTx.commit();
+            replayTx = null; // Mark as committed so we don't rollback in catch block
+            log.info("ES transaction commit completed in {}ms", System.currentTimeMillis() - commitStartTime);
+
+            long totalTime = System.currentTimeMillis() - startTime;
+            log.info("Successfully replayed mutation for index: {}. Total time: {}ms", entry.getIndexName(), totalTime);
+
+        } catch (TemporaryBackendException e) {
+            // Already a TemporaryBackendException from JanusGraph - rethrow as-is
+            log.warn("Temporary backend exception replaying DLQ entry: {}", e.getMessage());
+            cleanupFailedTransactions(replayTx, standardJanusGraphTx);
+            throw e;
+        } catch (IOException e) {
+            // JSON deserialization error - permanent failure
+            log.error("Failed to deserialize DLQ entry JSON - permanent failure", e);
+            cleanupFailedTransactions(replayTx, standardJanusGraphTx);
+            throw e;
+        } catch (Exception e) {
+            // Other exceptions - might be permanent (bad data, schema issues, etc.)
+            log.error("Error replaying DLQ entry - treating as permanent failure", e);
+            cleanupFailedTransactions(replayTx, standardJanusGraphTx);
+            throw e;
+        }
+    }
+
+    /**
+     * Cleanup transactions when replay fails
+     */
+    private void cleanupFailedTransactions(BaseTransaction replayTx, StandardJanusGraphTx standardJanusGraphTx) {
+        // Rollback ES transaction
+        if (replayTx != null) {
+            try {
+                replayTx.rollback();
+                log.debug("Rolled back ES transaction");
+            } catch (Exception rollbackException) {
+                log.error("Failed to rollback ES transaction", rollbackException);
+            }
+        }
+
+        // Rollback JanusGraph transaction (don't commit on failure!)
+        if (standardJanusGraphTx != null) {
+            try {
+                standardJanusGraphTx.rollback();
+                log.debug("Rolled back JanusGraph transaction");
+            } catch (Exception rollbackException) {
+                log.error("Failed to rollback JanusGraph transaction", rollbackException);
+            }
+        }
+    }
+
+    /**
+     * Reconstruct IndexMutation objects from serialized form
+     */
+    private Map<String, Map<String, IndexMutation>> reconstructMutations(DLQEntry entry, KeyInformation.IndexRetriever indexRetriever) {
+        Map<String, Map<String, IndexMutation>> result = new HashMap<>();
+
+        for (Map.Entry<String, Map<String, SerializableIndexMutation>> storeEntry :
+                entry.getMutations().entrySet()) {
+
+            String storeName = storeEntry.getKey();
+            Map<String, IndexMutation> storeMutations = new HashMap<>();
+
+            for (Map.Entry<String, SerializableIndexMutation> docEntry :
+                    storeEntry.getValue().entrySet()) {
+
+                String docId = docEntry.getKey();
+                SerializableIndexMutation serMut = docEntry.getValue();
+
+                // Reconstruct IndexMutation
+                IndexMutation mutation = new IndexMutation(
+                        indexRetriever.get(storeName),
+                        serMut.isNew(),
+                        serMut.isDeleted()
+                );
+
+                // Add additions
+                for (SerializableIndexMutation.SerializableIndexEntry add : serMut.getAdditions()) {
+                    mutation.addition(new IndexEntry(add.getField(), add.getValue()));
+                }
+
+                // Add deletions
+                for (SerializableIndexMutation.SerializableIndexEntry del : serMut.getDeletions()) {
+                    mutation.deletion(new IndexEntry(del.getField(), del.getValue()));
+                }
+
+                storeMutations.put(docId, mutation);
+            }
+
+            result.put(storeName, storeMutations);
+        }
+
+        return result;
+    }
+
+    /**
+     * Calculate exponential backoff delay for a failed message
+     * @param retryKey The partition-offset key
+     * @return The delay in milliseconds
+     */
+    private long calculateExponentialBackoff(String retryKey) {
+        long currentDelay = backoffTracker.getOrDefault(retryKey, (long) exponentialBackoffBaseDelayMs);
+        long nextDelay = (long) (currentDelay * exponentialBackoffMultiplier);
+
+        // Cap at maximum delay
+        nextDelay = Math.min(nextDelay, exponentialBackoffMaxDelayMs);
+
+        // Store for next time
+        backoffTracker.put(retryKey, nextDelay);
+
+        return currentDelay; // Return current delay, store next delay for future use
+    }
+
+    /**
+     * Reset exponential backoff for a successfully processed message
+     * @param retryKey The partition-offset key
+     */
+    private void resetExponentialBackoff(String retryKey) {
+        backoffTracker.remove(retryKey);
+    }
+
+    /**
+     * Health check for liveness/readiness probes
+     * @return true if the replay thread is healthy and running
+     */
+    public boolean isHealthy() {
+        // Check if thread is alive and healthy
+        boolean threadAlive = replayThread != null && replayThread.isAlive();
+        return isHealthy.get() && threadAlive;
+    }
+
+    /**
+     * Get replay status
+     */
+    public Map<String, Object> getStatus() {
+        Map<String, Object> status = new HashMap<>();
+        status.put("isRunning", isRunning.get());
+        status.put("isHealthy", isHealthy());
+        status.put("threadAlive", replayThread != null && replayThread.isAlive());
+        status.put("processedCount", processedCount.get());
+        status.put("errorCount", errorCount.get());
+        status.put("skippedCount", skippedCount.get());
+        status.put("topic", dlqTopic);
+        status.put("consumerGroup", consumerGroupId);
+        status.put("maxRetries", maxRetries);
+        status.put("activeRetries", retryTracker.size());
+        status.put("activeBackoffs", backoffTracker.size());
+
+        // Exponential backoff configuration
+        Map<String, Object> backoffConfig = new HashMap<>();
+        backoffConfig.put("baseDelayMs", exponentialBackoffBaseDelayMs);
+        backoffConfig.put("maxDelayMs", exponentialBackoffMaxDelayMs);
+        backoffConfig.put("multiplier", exponentialBackoffMultiplier);
+        status.put("exponentialBackoffConfig", backoffConfig);
+
+        return status;
+    }
+
+    private ObjectMapper configureMapper() {
+        ObjectMapper mapper = new ObjectMapper();
+        // Configure to handle property name differences
+        mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+        // Add custom deserializer for SerializableIndexMutation
+        SimpleModule module = new SimpleModule();
+        module.addDeserializer(SerializableIndexMutation.class, new JsonDeserializer<>() {
+            @Override
+            public SerializableIndexMutation deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
+                JsonNode node = p.getCodec().readTree(p);
+
+                // Handle both "new" and "isNew" fields
+                boolean isNew = node.has("new") ? node.get("new").asBoolean() :
+                        node.has("isNew") ? node.get("isNew").asBoolean() : false;
+
+                boolean isDeleted = node.has("isDeleted") ? node.get("isDeleted").asBoolean() : false;
+
+                List<SerializableIndexMutation.SerializableIndexEntry> additions = new ArrayList<>();
+                List<SerializableIndexMutation.SerializableIndexEntry> deletions = new ArrayList<>();
+
+                if (node.has("additions") && node.get("additions").isArray()) {
+                    for (JsonNode entry : node.get("additions")) {
+                        additions.add(new SerializableIndexMutation.SerializableIndexEntry(
+                                entry.get("field").asText(),
+                                mapper.treeToValue(entry.get("value"), Object.class)
+                        ));
+                    }
+                }
+
+                if (node.has("deletions") && node.get("deletions").isArray()) {
+                    for (JsonNode entry : node.get("deletions")) {
+                        deletions.add(new SerializableIndexMutation.SerializableIndexEntry(
+                                entry.get("field").asText(),
+                                mapper.treeToValue(entry.get("value"), Object.class)
+                        ));
+                    }
+                }
+
+                return new SerializableIndexMutation(isNew, isDeleted, additions, deletions);
+            }
+        });
+        mapper.registerModule(module);
+        return mapper;
+    }
+
+}
\ No newline at end of file
diff --git a/webapp/src/test/java/org/apache/atlas/web/service/DLQReplayServiceIntegrationTest.java b/webapp/src/test/java/org/apache/atlas/web/service/DLQReplayServiceIntegrationTest.java
new file mode 100644
index 00000000000..e40fa336d88
--- /dev/null
+++ b/webapp/src/test/java/org/apache/atlas/web/service/DLQReplayServiceIntegrationTest.java
@@ -0,0 +1,414 @@
+package org.apache.atlas.web.service;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.atlas.ApplicationProperties;
+import org.apache.atlas.AtlasException;
+import org.apache.atlas.repository.graphdb.janus.AtlasJanusGraph;
+import org.apache.kafka.clients.consumer.*;
+import org.apache.kafka.common.TopicPartition;
+import org.janusgraph.diskstorage.Backend;
+import org.janusgraph.diskstorage.BackendException;
+import org.janusgraph.diskstorage.BaseTransaction;
+import org.janusgraph.diskstorage.BaseTransactionConfig;
+import org.janusgraph.diskstorage.BaseTransactionConfigurable;
+import org.janusgraph.diskstorage.configuration.Configuration;
+import org.janusgraph.diskstorage.es.ElasticSearchIndex;
+import org.janusgraph.diskstorage.indexing.KeyInformation;
+import org.janusgraph.diskstorage.keycolumnvalue.StoreFeatures;
+import org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration;
+import org.janusgraph.graphdb.database.IndexSerializer;
+import org.janusgraph.graphdb.database.StandardJanusGraph;
+import org.janusgraph.graphdb.database.index.IndexInfoRetriever;
+import org.janusgraph.graphdb.transaction.StandardJanusGraphTx;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * Integration tests for DLQReplayService focusing on end-to-end scenarios
+ */
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class DLQReplayServiceIntegrationTest {
+
+    @Mock
+    private AtlasJanusGraph atlasJanusGraph;
+    @Mock
+    private StandardJanusGraph standardJanusGraph;
+    @Mock
+    private GraphDatabaseConfiguration graphConfig;
+    @Mock
+    private Configuration configuration;
+    @Mock
+    private ElasticSearchIndex esIndex;
+    @Mock
+    private IndexSerializer indexSerializer;
+    @Mock
+    private StoreFeatures storeFeatures;
+    @Mock
+    private StandardJanusGraphTx standardJanusGraphTx;
+    @Mock
+    private IndexInfoRetriever indexInfoRetriever;
+    @Mock
+    private KeyInformation.IndexRetriever keyInformationRetriever;
+    @Mock
+    private KeyInformation.StoreRetriever storeRetriever;
+    @Mock
+    private BaseTransactionConfigurable esTransaction;
+    @Mock
+    private KafkaConsumer<String, String> mockConsumer;
+
+    private DLQReplayService dlqReplayService;
+    
+    private MockedStatic<ApplicationProperties> mockApplicationProperties;
+    private MockedStatic<Backend> mockBackend;
+
+    @BeforeEach
+    void setUp() throws AtlasException {
+        // Clean up any existing static mocks first (defensive programming)
+        cleanupStaticMocks();
+        
+        try {
+            setupMocks();
+        } catch (BackendException e) {
+            throw new RuntimeException(e);
+        }
+        
+        // Mock ApplicationProperties (static)
+        //mockApplicationProperties = mockStatic(ApplicationProperties.class);
+        org.apache.commons.configuration.Configuration mockConfig = mock(org.apache.commons.configuration.Configuration.class);
+        mockApplicationProperties.when(ApplicationProperties::get).thenReturn(mockConfig);
+        when(mockConfig.getString("atlas.graph.kafka.bootstrap.servers")).thenReturn("localhost:9092");
+
+        // Mock Backend.getImplementationClass to return our mocked esIndex (static)
+        mockBackend = mockStatic(Backend.class);
+        mockBackend.when(() -> Backend.getImplementationClass(any(), anyString(), any())).thenReturn(esIndex);
+
+        dlqReplayService = spy(new DLQReplayService(atlasJanusGraph));
+        
+        // Inject mocks
+        ReflectionTestUtils.setField(dlqReplayService, "esIndex", esIndex);
+        ReflectionTestUtils.setField(dlqReplayService, "indexSerializer", indexSerializer);
+        ReflectionTestUtils.setField(dlqReplayService, "standardJanusGraph", standardJanusGraph);
+        ReflectionTestUtils.setField(dlqReplayService, "consumer", mockConsumer);
+        
+        // Set test configuration
+        ReflectionTestUtils.setField(dlqReplayService, "maxRetries", 3);
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffBaseDelayMs", 100);
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffMaxDelayMs", 1000);
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffMultiplier", 2.0);
+        ReflectionTestUtils.setField(dlqReplayService, "errorBackoffMs", 500);
+        
+        // Clear internal tracking maps to ensure test isolation
+        Map<String, Integer> retryTracker = (Map<String, Integer>) ReflectionTestUtils.getField(dlqReplayService, "retryTracker");
+        Map<String, Long> backoffTracker = (Map<String, Long>) ReflectionTestUtils.getField(dlqReplayService, "backoffTracker");
+        if (retryTracker != null) {
+            retryTracker.clear();
+        }
+        if (backoffTracker != null) {
+            backoffTracker.clear();
+        }
+    }
+    
+    @AfterEach
+    void tearDown() {
+        cleanupStaticMocks();
+    }
+    
+    private void cleanupStaticMocks() {
+        // Close static mocks to prevent "already registered" errors
+        if (mockApplicationProperties != null) {
+            try {
+                mockApplicationProperties.close();
+            } catch (Exception e) {
+                // Ignore - mock might already be closed
+            }
+            mockApplicationProperties = null;
+        }
+        if (mockBackend != null) {
+            try {
+                mockBackend.close();
+            } catch (Exception e) {
+                // Ignore - mock might already be closed
+            }
+            mockBackend = null;
+        }
+    }
+
+    private void setupMocks() throws AtlasException, BackendException {
+        when(atlasJanusGraph.getGraph()).thenReturn(standardJanusGraph);
+        when(standardJanusGraph.getConfiguration()).thenReturn(graphConfig);
+        when(graphConfig.getConfiguration()).thenReturn(configuration);
+        when(graphConfig.getBackend()).thenReturn(mock(org.janusgraph.diskstorage.Backend.class));
+        when(graphConfig.getBackend().getStoreFeatures()).thenReturn(storeFeatures);
+        when(graphConfig.getSerializer()).thenReturn(mock(org.janusgraph.graphdb.database.serialize.Serializer.class));
+        when(graphConfig.getBackend().getIndexInformation()).thenReturn(ImmutableMap.of("search", mock(org.janusgraph.diskstorage.indexing.IndexInformation.class)));
+        when(storeFeatures.isDistributed()).thenReturn(true);
+        when(storeFeatures.isKeyOrdered()).thenReturn(true);
+        when(configuration.restrictTo(anyString())).thenReturn(configuration);
+        when(configuration.get(any(), anyString())).thenReturn("elasticsearch");
+
+        mockApplicationProperties = mockStatic(ApplicationProperties.class);
+        org.apache.commons.configuration.Configuration mockConfig = mock(org.apache.commons.configuration.Configuration.class);
+        when(ApplicationProperties.get()).thenReturn(mockConfig);
+        when(mockConfig.getString("atlas.graph.kafka.bootstrap.servers")).thenReturn("localhost:9092");
+        
+        // Setup JanusGraph transaction mocks
+        when(standardJanusGraph.newTransaction()).thenReturn(standardJanusGraphTx);
+        when(indexSerializer.getIndexInfoRetriever(any())).thenReturn(indexInfoRetriever);
+        when(indexInfoRetriever.get(anyString())).thenReturn(keyInformationRetriever);
+        when(keyInformationRetriever.get(anyString())).thenReturn(storeRetriever);
+        when(esIndex.beginTransaction(any(BaseTransactionConfig.class))).thenReturn(esTransaction);
+    }
+
+    @Test
+    void testPoisonPill_SkippedAfterMaxRetries() {
+        // Arrange
+        String retryKey = "0-100";
+        Map<String, Integer> retryTracker = new ConcurrentHashMap<>();
+        ReflectionTestUtils.setField(dlqReplayService, "retryTracker", retryTracker);
+        
+        // Act - Simulate 3 failures (max retries)
+        retryTracker.put(retryKey, 1);
+        retryTracker.put(retryKey, 2);
+        retryTracker.put(retryKey, 3);
+        
+        int finalCount = retryTracker.get(retryKey);
+        
+        // Assert
+        assertEquals(3, finalCount);
+        assertTrue(finalCount >= 3, "Should reach maxRetries and be skipped");
+    }
+
+    @Test
+    void testBackoffProgression_MultipleMessages() {
+        // Arrange
+        String message1Key = "0-100";
+        String message2Key = "0-101";
+        
+        // Act - Build up backoff for message 1
+        long delay1_1 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", message1Key);
+        long delay1_2 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", message1Key);
+        long delay1_3 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", message1Key);
+        
+        // Act - Message 2 should start fresh
+        long delay2_1 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", message2Key);
+        
+        // Assert - Message 1 progresses
+        assertEquals(100, delay1_1);
+        assertEquals(200, delay1_2);
+        assertEquals(400, delay1_3);
+        
+        // Assert - Message 2 starts from base
+        assertEquals(100, delay2_1, "Different message should start from base delay");
+    }
+
+    @Test
+    void testBackoffReset_OnSuccessfulProcessing() {
+        // Arrange
+        String retryKey = "0-100";
+        
+        // Act - Build up delay
+        long delay1 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        long delay2 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        long delay3 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        
+        // Verify progression
+        assertEquals(100, delay1);
+        assertEquals(200, delay2);
+        assertEquals(400, delay3);
+        
+        // Act - Reset on success
+        ReflectionTestUtils.invokeMethod(dlqReplayService, "resetExponentialBackoff", retryKey);
+        
+        // Act - Next attempt should be back to base
+        long delayAfterReset = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        
+        // Assert
+        assertEquals(100, delayAfterReset, "Should reset to base delay after success");
+    }
+
+    @Test
+    void testPartitionRevocation_CleansUpTrackers() {
+        // Arrange
+        Map<String, Integer> retryTracker = new ConcurrentHashMap<>();
+        Map<String, Long> backoffTracker = new ConcurrentHashMap<>();
+        
+        retryTracker.put("0-100", 2);
+        retryTracker.put("0-101", 1);
+        retryTracker.put("1-200", 3);
+        
+        backoffTracker.put("0-100", 4000L);
+        backoffTracker.put("0-101", 2000L);
+        backoffTracker.put("1-200", 8000L);
+        
+        ReflectionTestUtils.setField(dlqReplayService, "retryTracker", retryTracker);
+        ReflectionTestUtils.setField(dlqReplayService, "backoffTracker", backoffTracker);
+        
+        // Act - Simulate partition 0 revocation
+        Collection<TopicPartition> revokedPartitions = Collections.singletonList(
+            new TopicPartition("TEST_DLQ", 0)
+        );
+        
+        // Manually call cleanup logic
+        String partitionPrefix = "0-";
+        retryTracker.keySet().removeIf(key -> key.startsWith(partitionPrefix));
+        backoffTracker.keySet().removeIf(key -> key.startsWith(partitionPrefix));
+        
+        // Assert
+        assertFalse(retryTracker.containsKey("0-100"), "Partition 0 entries should be removed");
+        assertFalse(retryTracker.containsKey("0-101"), "Partition 0 entries should be removed");
+        assertTrue(retryTracker.containsKey("1-200"), "Other partition entries should remain");
+        
+        assertFalse(backoffTracker.containsKey("0-100"), "Partition 0 backoff should be removed");
+        assertFalse(backoffTracker.containsKey("0-101"), "Partition 0 backoff should be removed");
+        assertTrue(backoffTracker.containsKey("1-200"), "Other partition backoff should remain");
+    }
+
+    @Test
+    void testMetricsTracking_ProcessedCount() {
+        // Arrange
+        AtomicInteger processedCount = new AtomicInteger(0);
+        AtomicInteger errorCount = new AtomicInteger(0);
+        AtomicInteger skippedCount = new AtomicInteger(0);
+        
+        ReflectionTestUtils.setField(dlqReplayService, "processedCount", processedCount);
+        ReflectionTestUtils.setField(dlqReplayService, "errorCount", errorCount);
+        ReflectionTestUtils.setField(dlqReplayService, "skippedCount", skippedCount);
+        
+        // Act - Simulate processing
+        processedCount.incrementAndGet();
+        processedCount.incrementAndGet();
+        errorCount.incrementAndGet();
+        skippedCount.incrementAndGet();
+        
+        // Assert
+        assertEquals(2, processedCount.get());
+        assertEquals(1, errorCount.get());
+        assertEquals(1, skippedCount.get());
+    }
+
+    @Test
+    void testStatusEndpoint_ReturnsCompleteInformation() {
+        // Arrange
+        AtomicInteger processedCount = new AtomicInteger(150);
+        AtomicInteger errorCount = new AtomicInteger(10);
+        AtomicInteger skippedCount = new AtomicInteger(3);
+        
+        Map<String, Integer> retryTracker = new ConcurrentHashMap<>();
+        retryTracker.put("0-100", 2);
+        retryTracker.put("0-101", 1);
+        
+        Map<String, Long> backoffTracker = new ConcurrentHashMap<>();
+        backoffTracker.put("0-100", 4000L);
+        
+        ReflectionTestUtils.setField(dlqReplayService, "processedCount", processedCount);
+        ReflectionTestUtils.setField(dlqReplayService, "errorCount", errorCount);
+        ReflectionTestUtils.setField(dlqReplayService, "skippedCount", skippedCount);
+        ReflectionTestUtils.setField(dlqReplayService, "retryTracker", retryTracker);
+        ReflectionTestUtils.setField(dlqReplayService, "backoffTracker", backoffTracker);
+        
+        // Act
+        Map<String, Object> status = dlqReplayService.getStatus();
+        
+        // Assert
+        assertEquals(150, status.get("processedCount"));
+        assertEquals(10, status.get("errorCount"));
+        assertEquals(3, status.get("skippedCount"));
+        assertEquals(2, status.get("activeRetries"));
+        assertEquals(1, status.get("activeBackoffs"));
+        assertNotNull(status.get("exponentialBackoffConfig"));
+    }
+
+    @Test
+    void testTransactionCleanup_OnFailure() throws Exception {
+        // Arrange
+        BaseTransaction esTransaction = mock(BaseTransaction.class);
+        StandardJanusGraphTx janusTransaction = mock(StandardJanusGraphTx.class);
+        
+        // Act
+        ReflectionTestUtils.invokeMethod(dlqReplayService, "cleanupFailedTransactions", 
+            esTransaction, janusTransaction);
+        
+        // Assert
+        verify(esTransaction).rollback();
+        verify(janusTransaction).rollback();
+    }
+
+    @Test
+    void testExponentialBackoff_DifferentMultipliers() {
+        // Test with multiplier 1.5
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffMultiplier", 1.5);
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffBaseDelayMs", 1000);
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffMaxDelayMs", 5000);
+        
+        String retryKey = "0-100";
+        
+        long delay1 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        long delay2 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        long delay3 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        
+        assertEquals(1000, delay1, "First delay should be base");
+        assertEquals(1500, delay2, "Second delay should be 1000 * 1.5");
+        assertEquals(2250, delay3, "Third delay should be 1500 * 1.5");
+    }
+
+    @Test
+    void testRetryLimit_ReachedAfterExactAttempts() {
+        // Arrange
+        Map<String, Integer> retryTracker = new ConcurrentHashMap<>();
+        ReflectionTestUtils.setField(dlqReplayService, "retryTracker", retryTracker);
+        ReflectionTestUtils.setField(dlqReplayService, "maxRetries", 5);
+        
+        String retryKey = "0-100";
+        
+        // Act - Increment retry count 5 times
+        for (int i = 1; i <= 5; i++) {
+            retryTracker.put(retryKey, i);
+        }
+        
+        int finalCount = retryTracker.get(retryKey);
+        
+        // Assert
+        assertEquals(5, finalCount);
+        assertTrue(finalCount >= 5, "Should have reached max retries of 5");
+    }
+
+    @Test
+    void testBackoffCap_WithVeryHighMultiplier() {
+        // Arrange - Set extreme multiplier to test cap
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffMultiplier", 10.0);
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffBaseDelayMs", 100);
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffMaxDelayMs", 5000);
+        
+        String retryKey = "0-100";
+        
+        // Act - Multiple attempts
+        long delay1 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        long delay2 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        long delay3 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        long delay4 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        
+        // Assert
+        assertEquals(100, delay1);
+        assertEquals(1000, delay2);  // 100 * 10
+        assertEquals(5000, delay3);  // Capped at max (would be 10000)
+        assertEquals(5000, delay4);  // Stays at cap
+    }
+}
+
diff --git a/webapp/src/test/java/org/apache/atlas/web/service/DLQReplayServiceTest.java b/webapp/src/test/java/org/apache/atlas/web/service/DLQReplayServiceTest.java
new file mode 100644
index 00000000000..fcb39566d13
--- /dev/null
+++ b/webapp/src/test/java/org/apache/atlas/web/service/DLQReplayServiceTest.java
@@ -0,0 +1,511 @@
+package org.apache.atlas.web.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.atlas.ApplicationProperties;
+import org.apache.atlas.AtlasException;
+import org.apache.atlas.repository.graphdb.janus.AtlasJanusGraph;
+import org.janusgraph.diskstorage.Backend;
+import org.janusgraph.diskstorage.BaseTransaction;
+import org.janusgraph.diskstorage.BaseTransactionConfigurable;
+import org.janusgraph.diskstorage.TemporaryBackendException;
+import org.janusgraph.diskstorage.configuration.Configuration;
+import org.janusgraph.diskstorage.dlq.DLQEntry;
+import org.janusgraph.diskstorage.dlq.SerializableIndexMutation;
+import org.janusgraph.diskstorage.es.ElasticSearchIndex;
+import org.janusgraph.diskstorage.indexing.KeyInformation;
+import org.janusgraph.diskstorage.keycolumnvalue.StoreFeatures;
+import org.janusgraph.graphdb.configuration.GraphDatabaseConfiguration;
+import org.janusgraph.graphdb.database.IndexSerializer;
+import org.janusgraph.graphdb.database.StandardJanusGraph;
+import org.janusgraph.graphdb.database.index.IndexInfoRetriever;
+import org.janusgraph.graphdb.transaction.StandardJanusGraphTx;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.io.IOException;
+import java.util.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+@ExtendWith(MockitoExtension.class)
+class DLQReplayServiceTest {
+
+    @Mock
+    private AtlasJanusGraph atlasJanusGraph;
+    @Mock
+    private StandardJanusGraph standardJanusGraph;
+    @Mock
+    private GraphDatabaseConfiguration graphConfig;
+    @Mock
+    private Configuration configuration;
+    @Mock
+    private ElasticSearchIndex esIndex;
+    @Mock
+    private IndexSerializer indexSerializer;
+    @Mock
+    private StoreFeatures storeFeatures;
+    @Mock
+    private StandardJanusGraphTx standardJanusGraphTx;
+    @Mock
+    private IndexInfoRetriever indexInfoRetriever;
+    @Mock
+    private KeyInformation.IndexRetriever keyInformationRetriever;
+    @Mock
+    private KeyInformation.StoreRetriever storeRetriever;
+    @Mock
+    private BaseTransactionConfigurable esTransaction;
+
+    private DLQReplayService dlqReplayService;
+    private ObjectMapper objectMapper;
+    
+    private MockedStatic<ApplicationProperties> mockApplicationProperties;
+    private MockedStatic<Backend> mockBackend;
+
+    @BeforeEach
+    void setUp() throws AtlasException {
+        // Clean up any existing static mocks first (defensive programming)
+        cleanupStaticMocks();
+        
+        // Mock the graph hierarchy
+        when(atlasJanusGraph.getGraph()).thenReturn(standardJanusGraph);
+        when(standardJanusGraph.getConfiguration()).thenReturn(graphConfig);
+        when(graphConfig.getConfiguration()).thenReturn(configuration);
+        when(graphConfig.getBackend()).thenReturn(mock(org.janusgraph.diskstorage.Backend.class));
+        when(graphConfig.getBackend().getStoreFeatures()).thenReturn(storeFeatures);
+        when(graphConfig.getSerializer()).thenReturn(mock(org.janusgraph.graphdb.database.serialize.Serializer.class));
+        when(graphConfig.getBackend().getIndexInformation()).thenReturn(Map.of("search", mock(org.janusgraph.diskstorage.indexing.IndexInformation.class)));
+        when(storeFeatures.isDistributed()).thenReturn(true);
+        when(storeFeatures.isKeyOrdered()).thenReturn(true);
+
+        // Mock configuration values
+        when(configuration.restrictTo(anyString())).thenReturn(configuration);
+        when(configuration.get(any(), anyString())).thenReturn("elasticsearch");
+
+        // Mock ApplicationProperties (static)
+        mockApplicationProperties = mockStatic(ApplicationProperties.class);
+        org.apache.commons.configuration.Configuration mockConfig = mock(org.apache.commons.configuration.Configuration.class);
+        mockApplicationProperties.when(ApplicationProperties::get).thenReturn(mockConfig);
+        when(mockConfig.getString("atlas.graph.kafka.bootstrap.servers")).thenReturn("localhost:9092");
+
+        // Mock Backend.getImplementationClass to return our mocked esIndex (static)
+        mockBackend = mockStatic(Backend.class);
+        mockBackend.when(() -> Backend.getImplementationClass(any(), anyString(), any())).thenReturn(esIndex);
+
+        // Create service - now it will use mocked esIndex
+        dlqReplayService = spy(new DLQReplayService(atlasJanusGraph));
+        
+        // Inject remaining mocks
+        ReflectionTestUtils.setField(dlqReplayService, "indexSerializer", indexSerializer);
+        ReflectionTestUtils.setField(dlqReplayService, "standardJanusGraph", standardJanusGraph);
+        
+        // Set test configuration values
+        ReflectionTestUtils.setField(dlqReplayService, "dlqTopic", "TEST_DLQ");
+        ReflectionTestUtils.setField(dlqReplayService, "consumerGroupId", "test_group");
+        ReflectionTestUtils.setField(dlqReplayService, "maxRetries", 3);
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffBaseDelayMs", 100);
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffMaxDelayMs", 1000);
+        ReflectionTestUtils.setField(dlqReplayService, "exponentialBackoffMultiplier", 2.0);
+        
+        // Clear internal tracking maps to ensure test isolation
+        Map<String, Integer> retryTracker = (Map<String, Integer>) ReflectionTestUtils.getField(dlqReplayService, "retryTracker");
+        Map<String, Long> backoffTracker = (Map<String, Long>) ReflectionTestUtils.getField(dlqReplayService, "backoffTracker");
+        if (retryTracker != null) {
+            retryTracker.clear();
+        }
+        if (backoffTracker != null) {
+            backoffTracker.clear();
+        }
+        
+        objectMapper = new ObjectMapper();
+    }
+    
+    @AfterEach
+    void tearDown() {
+        cleanupStaticMocks();
+    }
+    
+    private void cleanupStaticMocks() {
+        // Close static mocks to prevent "already registered" errors
+        if (mockApplicationProperties != null) {
+            try {
+                mockApplicationProperties.close();
+            } catch (Exception e) {
+                // Ignore - mock might already be closed
+            }
+            mockApplicationProperties = null;
+        }
+        if (mockBackend != null) {
+            try {
+                mockBackend.close();
+            } catch (Exception e) {
+                // Ignore - mock might already be closed
+            }
+            mockBackend = null;
+        }
+    }
+
+    @Test
+    void testReplayDLQEntry_Success() throws Exception {
+        // Arrange
+        String dlqJson = createValidDLQJson();
+        
+        when(standardJanusGraph.newTransaction()).thenReturn(standardJanusGraphTx);
+        when(indexSerializer.getIndexInfoRetriever(any())).thenReturn(indexInfoRetriever);
+        when(indexInfoRetriever.get(anyString())).thenReturn(keyInformationRetriever);
+        when(keyInformationRetriever.get(anyString())).thenReturn(storeRetriever);
+        when(esIndex.beginTransaction(any())).thenReturn(esTransaction);
+        
+        // Act
+        ReflectionTestUtils.invokeMethod(dlqReplayService, "replayDLQEntry", dlqJson);
+        
+        // Assert
+        verify(esIndex).mutate(any(), any(), eq(esTransaction));
+        verify(esTransaction).commit();
+        verify(standardJanusGraphTx).commit();
+        verify(esTransaction, never()).rollback();
+        verify(standardJanusGraphTx, never()).rollback();
+    }
+
+    @Test
+    void testReplayDLQEntry_TemporaryBackendException() throws Exception {
+        // Arrange
+        String dlqJson = createValidDLQJson();
+        
+        when(standardJanusGraph.newTransaction()).thenReturn(standardJanusGraphTx);
+        when(indexSerializer.getIndexInfoRetriever(any())).thenReturn(indexInfoRetriever);
+        when(indexInfoRetriever.get(anyString())).thenReturn(keyInformationRetriever);
+        when(keyInformationRetriever.get(anyString())).thenReturn(storeRetriever);
+        when(esIndex.beginTransaction(any())).thenReturn(esTransaction);
+        
+        doThrow(new TemporaryBackendException("ES cluster block"))
+            .when(esIndex).mutate(any(), any(), eq(esTransaction));
+        
+        // Act & Assert
+        // ReflectionTestUtils wraps exceptions in UndeclaredThrowableException, so we need to unwrap
+        Exception thrown = assertThrows(Exception.class, () -> {
+            ReflectionTestUtils.invokeMethod(dlqReplayService, "replayDLQEntry", dlqJson);
+        });
+        
+        // Unwrap the exception
+        Throwable cause = thrown.getCause();
+        assertTrue(cause instanceof TemporaryBackendException, 
+            "Expected TemporaryBackendException but got: " + (cause != null ? cause.getClass() : "null"));
+        
+        verify(esTransaction).rollback();
+        verify(standardJanusGraphTx).rollback();
+        verify(esTransaction, never()).commit();
+        verify(standardJanusGraphTx, never()).commit();
+    }
+
+    @Test
+    void testReplayDLQEntry_PermanentException() throws Exception {
+        // Arrange
+        String dlqJson = createValidDLQJson();
+        
+        when(standardJanusGraph.newTransaction()).thenReturn(standardJanusGraphTx);
+        when(indexSerializer.getIndexInfoRetriever(any())).thenReturn(indexInfoRetriever);
+        when(indexInfoRetriever.get(anyString())).thenReturn(keyInformationRetriever);
+        when(keyInformationRetriever.get(anyString())).thenReturn(storeRetriever);
+        when(esIndex.beginTransaction(any())).thenReturn(esTransaction);
+        
+        doThrow(new RuntimeException("Schema mismatch"))
+            .when(esIndex).mutate(any(), any(), eq(esTransaction));
+        
+        // Act & Assert
+        assertThrows(RuntimeException.class, () -> {
+            ReflectionTestUtils.invokeMethod(dlqReplayService, "replayDLQEntry", dlqJson);
+        });
+        
+        verify(esTransaction).rollback();
+        verify(standardJanusGraphTx).rollback();
+    }
+
+    @Test
+    void testReplayDLQEntry_JsonParsingError() throws Exception {
+        // Arrange
+        String invalidJson = "{invalid json}";
+        
+        // Act & Assert
+        // ReflectionTestUtils wraps exceptions in UndeclaredThrowableException, so we need to unwrap
+        Exception thrown = assertThrows(Exception.class, () -> {
+            ReflectionTestUtils.invokeMethod(dlqReplayService, "replayDLQEntry", invalidJson);
+        });
+        
+        // Unwrap the exception
+        Throwable cause = thrown.getCause();
+        assertTrue(cause instanceof IOException, 
+            "Expected IOException but got: " + (cause != null ? cause.getClass() : "null"));
+        
+        verify(esIndex, never()).mutate(any(), any(), any());
+    }
+
+    @Test
+    void testExponentialBackoff_Progression() {
+        // Arrange
+        String retryKey = "0-100";
+        
+        // Act & Assert - Should follow exponential progression
+        long delay1 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        assertEquals(100, delay1, "First delay should be base delay");
+        
+        long delay2 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        assertEquals(200, delay2, "Second delay should be 100 * 2.0");
+        
+        long delay3 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        assertEquals(400, delay3, "Third delay should be 200 * 2.0");
+        
+        long delay4 = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        assertEquals(800, delay4, "Fourth delay should be 400 * 2.0");
+    }
+
+    @Test
+    void testExponentialBackoff_CappedAtMax() {
+        // Arrange
+        String retryKey = "0-100";
+        
+        // Act - Call enough times to exceed max
+        for (int i = 0; i < 10; i++) {
+            ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        }
+        
+        long finalDelay = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        
+        // Assert
+        assertEquals(1000, finalDelay, "Delay should be capped at maxDelayMs (1000)");
+    }
+
+    @Test
+    void testExponentialBackoff_ResetOnSuccess() {
+        // Arrange
+        String retryKey = "0-100";
+        
+        // Build up delay
+        ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        
+        // Act - Reset
+        ReflectionTestUtils.invokeMethod(dlqReplayService, "resetExponentialBackoff", retryKey);
+        
+        // Assert - Should be back to base delay
+        long delayAfterReset = ReflectionTestUtils.invokeMethod(dlqReplayService, "calculateExponentialBackoff", retryKey);
+        assertEquals(100, delayAfterReset, "After reset, delay should return to base delay");
+    }
+
+    @Test
+    void testRetryTracking_IncrementAndSkip() {
+        // Arrange
+        Map<String, Integer> retryTracker = new HashMap<>();
+        ReflectionTestUtils.setField(dlqReplayService, "retryTracker", retryTracker);
+        String retryKey = "0-100";
+        
+        // Act - Simulate 3 failures
+        retryTracker.put(retryKey, 1);
+        int count1 = retryTracker.get(retryKey);
+        
+        retryTracker.put(retryKey, 2);
+        int count2 = retryTracker.get(retryKey);
+        
+        retryTracker.put(retryKey, 3);
+        int count3 = retryTracker.get(retryKey);
+        
+        // Assert
+        assertEquals(1, count1);
+        assertEquals(2, count2);
+        assertEquals(3, count3);
+        assertTrue(count3 >= 3, "Should reach maxRetries");
+    }
+
+    @Test
+    void testGetStatus_AllFields() {
+        // Arrange
+        ReflectionTestUtils.setField(dlqReplayService, "processedCount", 
+            new java.util.concurrent.atomic.AtomicInteger(100));
+        ReflectionTestUtils.setField(dlqReplayService, "errorCount", 
+            new java.util.concurrent.atomic.AtomicInteger(5));
+        ReflectionTestUtils.setField(dlqReplayService, "skippedCount", 
+            new java.util.concurrent.atomic.AtomicInteger(2));
+        
+        Map<String, Integer> retryTracker = new HashMap<>();
+        retryTracker.put("0-100", 2);
+        ReflectionTestUtils.setField(dlqReplayService, "retryTracker", retryTracker);
+        
+        Map<String, Long> backoffTracker = new HashMap<>();
+        backoffTracker.put("0-100", 4000L);
+        ReflectionTestUtils.setField(dlqReplayService, "backoffTracker", backoffTracker);
+        
+        // Act
+        Map<String, Object> status = dlqReplayService.getStatus();
+        
+        // Assert
+        assertEquals(100, status.get("processedCount"));
+        assertEquals(5, status.get("errorCount"));
+        assertEquals(2, status.get("skippedCount"));
+        assertEquals(1, status.get("activeRetries"));
+        assertEquals(1, status.get("activeBackoffs"));
+        assertEquals("TEST_DLQ", status.get("topic"));
+        assertEquals("test_group", status.get("consumerGroup"));
+        assertEquals(3, status.get("maxRetries"));
+        
+        @SuppressWarnings("unchecked")
+        Map<String, Object> backoffConfig = (Map<String, Object>) status.get("exponentialBackoffConfig");
+        assertNotNull(backoffConfig);
+        assertEquals(100, backoffConfig.get("baseDelayMs"));
+        assertEquals(1000, backoffConfig.get("maxDelayMs"));
+        assertEquals(2.0, backoffConfig.get("multiplier"));
+    }
+
+    @Test
+    void testIsHealthy_WhenHealthy() {
+        // Arrange
+        ReflectionTestUtils.setField(dlqReplayService, "isHealthy", 
+            new java.util.concurrent.atomic.AtomicBoolean(true));
+        
+        Thread mockThread = mock(Thread.class);
+        when(mockThread.isAlive()).thenReturn(true);
+        ReflectionTestUtils.setField(dlqReplayService, "replayThread", mockThread);
+        
+        // Act
+        boolean healthy = dlqReplayService.isHealthy();
+        
+        // Assert
+        assertTrue(healthy, "Service should be healthy when flag is true and thread is alive");
+    }
+
+    @Test
+    void testIsHealthy_WhenUnhealthy() {
+        // Arrange
+        ReflectionTestUtils.setField(dlqReplayService, "isHealthy", 
+            new java.util.concurrent.atomic.AtomicBoolean(false));
+        
+        Thread mockThread = mock(Thread.class);
+        when(mockThread.isAlive()).thenReturn(true);
+        ReflectionTestUtils.setField(dlqReplayService, "replayThread", mockThread);
+        
+        // Act
+        boolean healthy = dlqReplayService.isHealthy();
+        
+        // Assert
+        assertFalse(healthy, "Service should be unhealthy when flag is false");
+    }
+
+    @Test
+    void testIsHealthy_WhenThreadDead() {
+        // Arrange
+        ReflectionTestUtils.setField(dlqReplayService, "isHealthy", 
+            new java.util.concurrent.atomic.AtomicBoolean(true));
+        
+        Thread mockThread = mock(Thread.class);
+        when(mockThread.isAlive()).thenReturn(false);
+        ReflectionTestUtils.setField(dlqReplayService, "replayThread", mockThread);
+        
+        // Act
+        boolean healthy = dlqReplayService.isHealthy();
+        
+        // Assert
+        assertFalse(healthy, "Service should be unhealthy when thread is dead");
+    }
+
+    @Test
+    void testReconstructMutations() throws Exception {
+        // Arrange
+        DLQEntry entry = createDLQEntry();
+        
+        when(keyInformationRetriever.get(anyString())).thenReturn(storeRetriever);
+        
+        // Act
+        Map<String, Map<String, org.janusgraph.diskstorage.indexing.IndexMutation>> result = 
+            ReflectionTestUtils.invokeMethod(dlqReplayService, "reconstructMutations", entry, keyInformationRetriever);
+        
+        // Assert
+        assertNotNull(result);
+        assertTrue(result.containsKey("vertex_index"));
+        Map<String, org.janusgraph.diskstorage.indexing.IndexMutation> storeMutations = result.get("vertex_index");
+        assertNotNull(storeMutations);
+        assertTrue(storeMutations.containsKey("doc1"));
+    }
+
+    @Test
+    void testCleanupFailedTransactions_BothTransactionsNull() {
+        // Act & Assert - Should not throw
+        assertDoesNotThrow(() -> {
+            ReflectionTestUtils.invokeMethod(dlqReplayService, "cleanupFailedTransactions", null, null);
+        });
+    }
+
+    @Test
+    void testCleanupFailedTransactions_ESTransactionRollback() throws Exception {
+        // Arrange
+        BaseTransaction mockESTransaction = mock(BaseTransaction.class);
+        
+        // Act
+        ReflectionTestUtils.invokeMethod(dlqReplayService, "cleanupFailedTransactions", 
+            mockESTransaction, null);
+        
+        // Assert
+        verify(mockESTransaction).rollback();
+    }
+
+    @Test
+    void testCleanupFailedTransactions_JanusGraphTransactionRollback() throws Exception {
+        // Arrange
+        StandardJanusGraphTx mockJanusTx = mock(StandardJanusGraphTx.class);
+        
+        // Act
+        ReflectionTestUtils.invokeMethod(dlqReplayService, "cleanupFailedTransactions", 
+            null, mockJanusTx);
+        
+        // Assert
+        verify(mockJanusTx).rollback();
+    }
+
+    @Test
+    void testCleanupFailedTransactions_RollbackException() throws Exception {
+        // Arrange
+        BaseTransaction mockESTransaction = mock(BaseTransaction.class);
+        doThrow(new RuntimeException("Rollback failed")).when(mockESTransaction).rollback();
+        
+        // Act & Assert - Should log error but not throw
+        assertDoesNotThrow(() -> {
+            ReflectionTestUtils.invokeMethod(dlqReplayService, "cleanupFailedTransactions", 
+                mockESTransaction, null);
+        });
+        
+        verify(mockESTransaction).rollback();
+    }
+
+    // Helper methods
+
+    private String createValidDLQJson() throws Exception {
+        DLQEntry entry = createDLQEntry();
+        return objectMapper.writeValueAsString(entry);
+    }
+
+    private DLQEntry createDLQEntry() {
+        Map<String, Map<String, SerializableIndexMutation>> mutations = new HashMap<>();
+        Map<String, SerializableIndexMutation> storeMutations = new HashMap<>();
+        
+        List<SerializableIndexMutation.SerializableIndexEntry> additions = new ArrayList<>();
+        additions.add(new SerializableIndexMutation.SerializableIndexEntry("field1", "value1"));
+        
+        List<SerializableIndexMutation.SerializableIndexEntry> deletions = new ArrayList<>();
+        
+        SerializableIndexMutation mutation = new SerializableIndexMutation(
+            true, false, additions, deletions
+        );
+        
+        storeMutations.put("doc1", mutation);
+        mutations.put("vertex_index", storeMutations);
+        
+        return new DLQEntry("janusgraph", "vertex_index", mutations, "", null, 0L, null);
+    }
+}
+