AEM Assets Content Hub: Fixing Move Listener Timeout Critical Issue

When implementing custom asset processing in AEM Assets Content Hub, a common pattern is using a ResourceChangeListener to move uploaded images and enhance their metadata. However, this approach frequently leads to the listener being killed due to timeout—especially when performing multiple heavy metadata operations. Here's how to solve it using Sling Jobs for asynchronous processing.

Understanding the Problem

What is AEM Assets Content Hub?

AEM Assets Content Hub is a centralized portal within AEM as a Cloud Service that allows marketing teams, agencies, and distributed stakeholders to upload, discover, and distribute approved brand assets. It provides a streamlined interface for:

• Self-service uploads — Non-technical users can upload assets directly
• Asset discovery — Search and filter approved assets
• Distribution — Share assets via links or direct downloads
• Governance — Controlled access to brand-approved content

The Custom Processing Requirement

Many organizations need custom logic when assets are uploaded via Content Hub:

• Move assets to proper folder structures based on metadata
• Enhance metadata with auto-tagging, AI descriptions, or taxonomy mappings
• Trigger workflows for approval or processing
• Call external APIs for enrichment (DAM systems, PIMs, AI services)

The Synchronous Listener Problem

A typical implementation uses a ResourceChangeListener to detect uploads and process them immediately:

@Component(
    service = ResourceChangeListener.class,
    property = {
        ResourceChangeListener.PATHS + "=/content/dam/content-hub-uploads",
        ResourceChangeListener.CHANGES + "=ADDED"
    }
)
public class ContentHubAssetListener implements ResourceChangeListener {
    
    @Reference
    private ResourceResolverFactory resolverFactory;
    
    @Reference
    private AssetManager assetManager;
    
    @Override
    public void onChange(List<ResourceChange> changes) {
        try (ResourceResolver resolver = getServiceResolver()) {
            for (ResourceChange change : changes) {
                String path = change.getPath();
                
                // PROBLEM: These operations are HEAVY and SYNCHRONOUS
                moveAssetToTargetFolder(resolver, path);      // ~500ms
                extractAndEnhanceMetadata(resolver, path);    // ~2000ms
                callExternalAITagging(resolver, path);        // ~3000ms
                updateTaxonomyMappings(resolver, path);       // ~800ms
                triggerApprovalWorkflow(resolver, path);      // ~500ms
                
                resolver.commit();
            }
        } catch (Exception e) {
            log.error("Asset processing failed", e);
        }
    }
}

Symptoms of Timeout Issues

You'll see these signs in your logs and behavior:

# In error.log
*WARN* [oak-observation-1] org.apache.jackrabbit.oak.plugins.observation.ChangeProcessor 
    Listener exceeded timeout: com.example.ContentHubAssetListener - killing listener

*ERROR* [oak-observation-1] org.apache.sling.jcr.resource.internal.helper.jcr.JcrResourceListener 
    ResourceChange listener timed out after 5000ms

# Observable symptoms:
- Assets stuck in upload folder (never moved)
- Partial metadata updates (some fields set, others missing)
- Workflows never triggered
- Inconsistent state across batch uploads

The Solution: Sling Jobs for Asynchronous Processing

The fix is to make the listener thin—it should only detect changes and queue work for asynchronous processing. Sling Jobs provide guaranteed execution, retry logic, and proper threading.

                    SYNCHRONOUS (Problem)
┌─────────────────────────────────────────────────────────────────┐
│                                                                 │
│  Upload → Listener → Move → Metadata → AI → Taxonomy → Done    │
│                      ↑                                          │
│                      └── All on Oak thread (TIMEOUT!)           │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

                    ASYNCHRONOUS (Solution)
┌─────────────────────────────────────────────────────────────────┐
│                                                                 │
│  Upload → Listener → Create Job → Return immediately (~50ms)   │
│                          │                                      │
│                          ▼                                      │
│              ┌─────────────────────┐                            │
│              │   Sling Job Queue   │                            │
│              └──────────┬──────────┘                            │
│                         │                                       │
│                         ▼                                       │
│              Job Consumer (separate thread)                     │
│              Move → Metadata → AI → Taxonomy → Done             │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Architecture Components

1. Content Hub Upload — User uploads asset via Content Hub UI
2. Thin Listener — Detects change, creates job (~50ms)
3. Job Queue — Persisted, guaranteed delivery
4. Job Consumer — Heavy processing on separate thread (no timeout)

Implementation

Step 1: Define the Job Topic

Create a constants class to define your job topic:

package com.example.core.jobs;

public final class ContentHubJobConstants {
    
    // Job topic for asset processing
    public static final String TOPIC_ASSET_PROCESSING = 
        "com/example/contenthub/asset/processing";
    
    // Job properties
    public static final String PROP_ASSET_PATH = "assetPath";
    public static final String PROP_UPLOAD_USER = "uploadUser";
    public static final String PROP_UPLOAD_TIME = "uploadTime";
    public static final String PROP_SOURCE_FOLDER = "sourceFolder";
    
    private ContentHubJobConstants() {
        // Utility class
    }
}

Step 2: Create the Thin Listener (Job Producer)

The listener now only creates jobs—no heavy processing:

package com.example.core.listeners;

import org.apache.sling.api.resource.ResourceChange;
import org.apache.sling.api.resource.ResourceChangeListener;
import org.apache.sling.event.jobs.JobManager;
import org.osgi.service.component.annotations.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.example.core.jobs.ContentHubJobConstants.*;

@Component(
    service = ResourceChangeListener.class,
    immediate = true,
    property = {
        ResourceChangeListener.PATHS + "=/content/dam/content-hub-uploads",
        ResourceChangeListener.CHANGES + "=ADDED"
    }
)
public class ContentHubAssetListener implements ResourceChangeListener {
    
    private static final Logger log = LoggerFactory.getLogger(ContentHubAssetListener.class);
    
    @Reference
    private JobManager jobManager;
    
    @Override
    public void onChange(List<ResourceChange> changes) {
        for (ResourceChange change : changes) {
            String path = change.getPath();
            
            // Only process actual asset nodes (not subnodes like renditions)
            if (isAssetNode(path)) {
                createProcessingJob(path, change);
            }
        }
    }
    
    private boolean isAssetNode(String path) {
        // Filter to only process main asset nodes
        return path.matches(".*/content/dam/content-hub-uploads/[^/]+") 
            && !path.contains("/jcr:content")
            && !path.contains("/_jcr_content");
    }
    
    private void createProcessingJob(String assetPath, ResourceChange change) {
        Map<String, Object> jobProperties = new HashMap<>();
        jobProperties.put(PROP_ASSET_PATH, assetPath);
        jobProperties.put(PROP_UPLOAD_TIME, System.currentTimeMillis());
        jobProperties.put(PROP_SOURCE_FOLDER, "/content/dam/content-hub-uploads");
        
        // Add job to queue - returns immediately
        jobManager.addJob(TOPIC_ASSET_PROCESSING, jobProperties);
        
        log.info("Created processing job for Content Hub asset: {}", assetPath);
    }
}

Step 3: Create the Job Consumer (Heavy Processing)

The job consumer handles all the heavy lifting on a separate thread:

package com.example.core.jobs;

import com.day.cq.dam.api.Asset;
import com.day.cq.dam.api.AssetManager;
import org.apache.sling.api.resource.*;
import org.apache.sling.event.jobs.Job;
import org.apache.sling.event.jobs.consumer.JobConsumer;
import org.osgi.service.component.annotations.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jcr.Session;
import java.util.HashMap;
import java.util.Map;

import static com.example.core.jobs.ContentHubJobConstants.*;

@Component(
    service = JobConsumer.class,
    immediate = true,
    property = {
        JobConsumer.PROPERTY_TOPICS + "=" + TOPIC_ASSET_PROCESSING
    }
)
public class ContentHubAssetJobConsumer implements JobConsumer {
    
    private static final Logger log = LoggerFactory.getLogger(ContentHubAssetJobConsumer.class);
    
    private static final String SERVICE_USER = "content-hub-processor";
    
    @Reference
    private ResourceResolverFactory resolverFactory;
    
    @Reference
    private AssetMetadataEnhancer metadataEnhancer;
    
    @Reference
    private ExternalAIService aiService;
    
    @Reference
    private TaxonomyMapper taxonomyMapper;
    
    @Override
    public JobResult process(Job job) {
        String assetPath = job.getProperty(PROP_ASSET_PATH, String.class);
        
        if (assetPath == null) {
            log.error("Job missing asset path property");
            return JobResult.CANCEL; // Don't retry - invalid job
        }
        
        log.info("Processing Content Hub asset: {}", assetPath);
        
        try (ResourceResolver resolver = getServiceResolver()) {
            
            // Step 1: Validate asset still exists
            Resource assetResource = resolver.getResource(assetPath);
            if (assetResource == null) {
                log.warn("Asset no longer exists: {}", assetPath);
                return JobResult.OK; // Asset deleted - nothing to do
            }
            
            Asset asset = assetResource.adaptTo(Asset.class);
            if (asset == null) {
                log.error("Resource is not an asset: {}", assetPath);
                return JobResult.CANCEL;
            }
            
            // Step 2: Determine target folder based on metadata/rules
            String targetFolder = determineTargetFolder(asset);
            
            // Step 3: Move asset to target folder
            String newPath = moveAsset(resolver, assetPath, targetFolder);
            
            // Step 4: Enhance metadata (can take 2+ seconds)
            enhanceMetadata(resolver, newPath);
            
            // Step 5: Call external AI for smart tagging (can take 3+ seconds)
            applyAITags(resolver, newPath);
            
            // Step 6: Map to taxonomy (can take 1+ second)
            mapTaxonomy(resolver, newPath);
            
            // Step 7: Commit all changes
            resolver.commit();
            
            log.info("Successfully processed Content Hub asset: {} -> {}", assetPath, newPath);
            return JobResult.OK;
            
        } catch (PersistenceException e) {
            log.error("Failed to persist changes for asset: {}", assetPath, e);
            return JobResult.FAILED; // Will retry
            
        } catch (ExternalServiceException e) {
            log.warn("External service unavailable for asset: {}", assetPath, e);
            return JobResult.FAILED; // Will retry
            
        } catch (Exception e) {
            log.error("Unexpected error processing asset: {}", assetPath, e);
            return JobResult.CANCEL; // Don't retry - unknown error
        }
    }
    
    private ResourceResolver getServiceResolver() throws LoginException {
        Map<String, Object> params = new HashMap<>();
        params.put(ResourceResolverFactory.SUBSERVICE, SERVICE_USER);
        return resolverFactory.getServiceResourceResolver(params);
    }
    
    private String determineTargetFolder(Asset asset) {
        // Logic to determine target folder based on:
        // - Asset type (image, video, document)
        // - Metadata values (brand, campaign, region)
        // - File naming conventions
        // - Custom business rules
        
        String assetType = asset.getMimeType();
        String brand = asset.getMetadataValue("dam:brand");
        String campaign = asset.getMetadataValue("dam:campaign");
        
        if (brand != null && campaign != null) {
            return String.format("/content/dam/approved/%s/%s", brand, campaign);
        } else if (assetType != null && assetType.startsWith("image/")) {
            return "/content/dam/approved/images/uncategorized";
        } else {
            return "/content/dam/approved/other";
        }
    }
    
    private String moveAsset(ResourceResolver resolver, String sourcePath, String targetFolder) 
            throws PersistenceException {
        
        // Ensure target folder exists
        ensureFolderExists(resolver, targetFolder);
        
        // Get asset filename
        String filename = sourcePath.substring(sourcePath.lastIndexOf('/') + 1);
        String targetPath = targetFolder + "/" + filename;
        
        // Handle naming conflicts
        targetPath = resolveNamingConflict(resolver, targetPath);
        
        // Move the asset
        Resource source = resolver.getResource(sourcePath);
        resolver.move(sourcePath, targetPath.substring(0, targetPath.lastIndexOf('/')));
        
        log.info("Moved asset from {} to {}", sourcePath, targetPath);
        return targetPath;
    }
    
    private void ensureFolderExists(ResourceResolver resolver, String folderPath) 
            throws PersistenceException {
        
        Resource folder = resolver.getResource(folderPath);
        if (folder != null) {
            return;
        }
        
        // Create folder hierarchy
        String[] segments = folderPath.split("/");
        StringBuilder currentPath = new StringBuilder();
        
        for (String segment : segments) {
            if (segment.isEmpty()) continue;
            
            currentPath.append("/").append(segment);
            String path = currentPath.toString();
            
            if (resolver.getResource(path) == null) {
                Map<String, Object> props = new HashMap<>();
                props.put("jcr:primaryType", "sling:Folder");
                resolver.create(resolver.getResource(path.substring(0, path.lastIndexOf('/'))), 
                    segment, props);
            }
        }
    }
    
    private String resolveNamingConflict(ResourceResolver resolver, String targetPath) {
        if (resolver.getResource(targetPath) == null) {
            return targetPath;
        }
        
        // Add timestamp suffix for uniqueness
        String basePath = targetPath.substring(0, targetPath.lastIndexOf('.'));
        String extension = targetPath.substring(targetPath.lastIndexOf('.'));
        
        return basePath + "-" + System.currentTimeMillis() + extension;
    }
    
    private void enhanceMetadata(ResourceResolver resolver, String assetPath) {
        log.debug("Enhancing metadata for: {}", assetPath);
        
        Resource asset = resolver.getResource(assetPath);
        if (asset == null) return;
        
        // Use injected service for metadata enhancement
        metadataEnhancer.enhance(asset.adaptTo(Asset.class));
    }
    
    private void applyAITags(ResourceResolver resolver, String assetPath) 
            throws ExternalServiceException {
        log.debug("Applying AI tags for: {}", assetPath);
        
        Resource asset = resolver.getResource(assetPath);
        if (asset == null) return;
        
        // Call external AI service (Adobe Sensei, AWS Rekognition, etc.)
        aiService.tagAsset(asset.adaptTo(Asset.class));
    }
    
    private void mapTaxonomy(ResourceResolver resolver, String assetPath) {
        log.debug("Mapping taxonomy for: {}", assetPath);
        
        Resource asset = resolver.getResource(assetPath);
        if (asset == null) return;
        
        // Map to organizational taxonomy
        taxonomyMapper.map(asset.adaptTo(Asset.class));
    }
}

Step 4: Configure the Service User

Create the service user mapping for the job consumer:

// ui.config/src/main/content/jcr_root/apps/example/osgiconfig/config.author/
// org.apache.sling.serviceusermapping.impl.ServiceUserMapperImpl.amended-contenthub.cfg.json

{
    "user.mapping": [
        "com.example.core:content-hub-processor=content-hub-service-user"
    ]
}

And ensure the service user has appropriate permissions:

<!-- repo-init script -->
create service user content-hub-service-user with path system/example

set ACL for content-hub-service-user
    allow jcr:read,jcr:write,jcr:modifyProperties on /content/dam
    allow jcr:read on /conf
    allow jcr:read on /apps
end

Step 5: Configure Job Queue

Optimize the job queue for your workload:

// org.apache.sling.event.jobs.QueueConfiguration-contenthub.cfg.json

{
    "queue.name": "Content Hub Asset Processing",
    "queue.topics": [
        "com/example/contenthub/asset/processing"
    ],
    "queue.type": "ORDERED",
    "queue.maxparallel": 3,
    "queue.retries": 5,
    "queue.retrydelay": 2000,
    "queue.priority": "NORM"
}

The Metadata Enhancer Service

Here's an example metadata enhancement service that would be too slow for synchronous processing:

package com.example.core.services.impl;

import com.day.cq.dam.api.Asset;
import com.example.core.services.AssetMetadataEnhancer;
import org.apache.sling.api.resource.ModifiableValueMap;
import org.apache.sling.api.resource.Resource;
import org.osgi.service.component.annotations.Component;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.InputStream;
import java.util.Calendar;

@Component(service = AssetMetadataEnhancer.class)
public class AssetMetadataEnhancerImpl implements AssetMetadataEnhancer {
    
    private static final Logger log = LoggerFactory.getLogger(AssetMetadataEnhancerImpl.class);
    
    @Override
    public void enhance(Asset asset) {
        if (asset == null) return;
        
        Resource metadataResource = asset.adaptTo(Resource.class)
            .getChild("jcr:content/metadata");
        
        if (metadataResource == null) return;
        
        ModifiableValueMap metadata = metadataResource.adaptTo(ModifiableValueMap.class);
        
        // Extract and set file-based metadata
        extractImageDimensions(asset, metadata);
        extractColorProfile(asset, metadata);
        calculateFileHash(asset, metadata);
        
        // Set processing metadata
        metadata.put("dam:processedByContentHub", true);
        metadata.put("dam:processingDate", Calendar.getInstance());
        metadata.put("dam:processingVersion", "1.0");
        
        // Business metadata enrichment
        enrichBusinessMetadata(asset, metadata);
        
        log.info("Enhanced metadata for asset: {}", asset.getPath());
    }
    
    private void extractImageDimensions(Asset asset, ModifiableValueMap metadata) {
        // This operation reads the binary and can take 500ms+
        try {
            String mimeType = asset.getMimeType();
            if (mimeType != null && mimeType.startsWith("image/")) {
                // Extract dimensions using ImageIO or similar
                // metadata.put("dam:imageWidth", width);
                // metadata.put("dam:imageHeight", height);
            }
        } catch (Exception e) {
            log.warn("Could not extract dimensions: {}", asset.getPath(), e);
        }
    }
    
    private void extractColorProfile(Asset asset, ModifiableValueMap metadata) {
        // Color profile extraction - can take 300ms+
        // metadata.put("dam:colorSpace", colorSpace);
        // metadata.put("dam:colorProfile", profileName);
    }
    
    private void calculateFileHash(Asset asset, ModifiableValueMap metadata) {
        // SHA-256 hash for deduplication - can take 1000ms+ for large files
        try (InputStream is = asset.getOriginal().getStream()) {
            // String hash = DigestUtils.sha256Hex(is);
            // metadata.put("dam:sha256Hash", hash);
        } catch (Exception e) {
            log.warn("Could not calculate hash: {}", asset.getPath(), e);
        }
    }
    
    private void enrichBusinessMetadata(Asset asset, ModifiableValueMap metadata) {
        // Apply business rules for metadata enrichment
        String filename = asset.getName().toLowerCase();
        
        // Auto-categorize based on naming conventions
        if (filename.contains("hero")) {
            metadata.put("dam:assetCategory", "hero-image");
        } else if (filename.contains("thumb")) {
            metadata.put("dam:assetCategory", "thumbnail");
        } else if (filename.contains("banner")) {
            metadata.put("dam:assetCategory", "banner");
        }
    }
}

Handling Edge Cases

Batch Uploads

When users upload multiple files simultaneously via Content Hub, each file triggers a separate job. The queue configuration handles this gracefully:

// In the listener, add batch tracking metadata
private void createProcessingJob(String assetPath, ResourceChange change) {
    Map<String, Object> jobProperties = new HashMap<>();
    jobProperties.put(PROP_ASSET_PATH, assetPath);
    jobProperties.put(PROP_UPLOAD_TIME, System.currentTimeMillis());
    
    // Add batch ID for tracking related uploads
    String batchId = generateBatchId();
    jobProperties.put("batchId", batchId);
    
    jobManager.addJob(TOPIC_ASSET_PROCESSING, jobProperties);
}

Idempotency

Jobs may be retried. Ensure your processing is idempotent:

private String moveAsset(ResourceResolver resolver, String sourcePath, String targetFolder) 
        throws PersistenceException {
    
    // Check if already moved (idempotency)
    Resource source = resolver.getResource(sourcePath);
    if (source == null) {
        // Asset might have been moved by a previous attempt
        // Check if it exists in target location
        String filename = sourcePath.substring(sourcePath.lastIndexOf('/') + 1);
        String expectedTarget = targetFolder + "/" + filename;
        
        if (resolver.getResource(expectedTarget) != null) {
            log.info("Asset already moved to: {}", expectedTarget);
            return expectedTarget;
        }
        
        throw new PersistenceException("Asset not found: " + sourcePath);
    }
    
    // Proceed with move...
}

Error Notifications

Notify administrators when jobs fail permanently:

@Override
public JobResult process(Job job) {
    int retryCount = job.getRetryCount();
    int maxRetries = 5;
    
    try {
        // ... processing logic ...
        return JobResult.OK;
        
    } catch (Exception e) {
        if (retryCount >= maxRetries - 1) {
            // This is the last retry - send notification
            notifyAdminOfFailure(job, e);
        }
        return JobResult.FAILED;
    }
}

private void notifyAdminOfFailure(Job job, Exception e) {
    String assetPath = job.getProperty(PROP_ASSET_PATH, String.class);
    
    // Send email, Slack notification, or log to monitoring system
    log.error("ALERT: Content Hub asset processing permanently failed: {} - {}", 
        assetPath, e.getMessage());
    
    // Optionally move asset to a "failed" folder for manual review
}

Monitoring and Debugging

Job Queue Dashboard

Monitor job queues via the Sling Jobs Console:

• URL: /system/console/slingevent
• View active, queued, and failed jobs
• Manually retry or cancel jobs
• See processing statistics

Custom Health Check

@Component(
    service = HealthCheck.class,
    property = {
        HealthCheck.NAME + "=Content Hub Processing Queue",
        HealthCheck.TAGS + "=contenthub",
        HealthCheck.MBEAN_NAME + "=contentHubProcessingQueue"
    }
)
public class ContentHubQueueHealthCheck implements HealthCheck {
    
    @Reference
    private JobManager jobManager;
    
    @Override
    public Result execute() {
        FormattingResultLog log = new FormattingResultLog();
        
        // Check queue statistics
        Statistics stats = jobManager.getStatistics();
        long queuedJobs = stats.getNumberOfQueuedJobs();
        long activeJobs = stats.getNumberOfActiveJobs();
        
        if (queuedJobs > 100) {
            log.warn("High queue backlog: {} jobs waiting", queuedJobs);
        } else if (queuedJobs > 500) {
            log.critical("Critical queue backlog: {} jobs waiting", queuedJobs);
        } else {
            log.info("Queue healthy: {} queued, {} active", queuedJobs, activeJobs);
        }
        
        return new Result(log);
    }
}

Performance Comparison

Best Practices Summary