The catalog synchronization job publishes frequently updates about its progress and synchronization speed. The Solr indexing job however does not offer a such feature, although it would be very beneficial to better understand indexing performances. I faced recently a situation where the indexing job was not scaling: the more items were indexed, the longer the job ran to index one item. Without insight on the indexing speed, it was hard to know where to start investigations. Moreover, it is in general hard to estimate the finishing time of a Solr indexing job. To address these issues, I enhanced SAP Commerce, so that the indexing job would regularly update the statusCode attribute of the active CronJobHistory item with a message similar to the following one:

9,456,049 item(s) indexed within 14m 23s (10,957 items/s), 0 error(s)

Like for the catalog synchronization job, it then makes it possible to track live the Solr indexing job progresses through Backoffice.

Approach

In a nutshell, the Solr indexing job fetches the primary keys of the items to index, splits them into batches and indexes each batch in parallel. The approach is to implement the IndexerListener and IndexerBatchListener interfaces to intercept the batch end events and report the number of items being indexed in a given period. The hard part is that the batches are not connected to the indexing job as they run in their own session and in a different thread. Moreover, the job is not exposed to the batches. The only information connecting the indexing job to the batches is the indexing operation identifier. The logic to implement is consequently the following:

• Intercept the indexing start event to store the job with its indexing operation identifier. Batches can then get access to the indexing job via the indexing operation identifier.
• Intercept the batch end event to count the number of indexed items and refresh the indexing speed.
• Intercept the indexing end event to publish the final indexing speed and remove any allocated resources like the mapping between the job and its indexing operation identifier.

Refreshing the indexing speed each time a batch finishes could be too much and slow down the system. Therefore, the proposed approach is to update only after a certain time interval (e.g. 1 minute like the catalog synchronization job).

Implementation

The most important aspect to focus on during the implementation was to measure as accurately as possible without slowing down the operations. Other said, precision can be sacrificed over performance. I therefore banned synchronization during the implementation and relied on the AtomicInteger and AtomicLong classes to prevent issues with concurrent updates. First, I needed a container holding the indexing statistics for an indexing job.

public class SolrIndexerCronJobStatistics {
    private final SolrIndexerCronJobModel cronJob;
    private final AtomicLong totalIndexedItems = new AtomicLong(0);
    private final AtomicInteger totalErrors = new AtomicInteger(0);
    private Date indexStartTime;
    private Date indexEndTime;

    SolrIndexerCronJobStatistics(final SolrIndexerCronJobModel cronJob) {
        this.cronJob = cronJob;
        setIndexStartTime(new Date());
    }

    public SolrIndexerCronJobModel getCronJob() {
        return cronJob;
    }

    public long addIndexedItems(final int additionalIndexedItems) {
        return this.totalIndexedItems.addAndGet(additionalIndexedItems);
    }

    public int addErrors(final int additionalErrors) {
        return this.totalErrors.addAndGet(additionalErrors);
    }

    public long getTotalIndexedItems() {
        return this.totalIndexedItems.get();
    }

    public int getTotalErrors() {
        return this.totalErrors.get();
    }

    public Date getIndexStartTime() {
        return indexStartTime;
    }

    public void setIndexStartTime(Date indexStartTime) {
        this.indexStartTime = indexStartTime;
    }

    public Date getIndexEndTime() {
        return indexEndTime;
    }

    public void setIndexEndTime(Date indexEndTime) {
        this.indexEndTime = indexEndTime;
    }
}

Then, the listener shall implement the logic described in the previous section. When a batch ends, the indexing job statistics are updated and if the time interval since the last refresh is elapsed, it updates the statusCode attribute of the active CronJobHistory item. Note that the modifiedTime attribute is updated, which traditionally is useless since SAP Commerce overwrites it when modelService.save() is executed. The reason for this odd line is to prevent parallel updates of the CronJobHistory item. Under high parallelism, batches will very likely ends at the same time and it can take couple milliseconds to save the CronJobHistory in the database and update the modifiedTime attribute.

public class SolrIndexerCronJobListener implements IndexerListener, IndexerBatchListener {
    private static final Logger LOG = LogManager.getLogger(SolrIndexerCronJobListener.class);

    private ModelService modelService;
    private SessionService sessionService;

    private final Map<Long, SolrIndexerCronJobStatistics> solrIndexerCronJobStatistics = new ConcurrentHashMap<>();
    private long updateInterval = 1000 * 60 * 1;

    @Override
    public void beforeBatch(final IndexerBatchContext ctx) throws IndexerException {
        // Nothing to do
    }

    @Override
    public void afterBatch(final IndexerBatchContext ctx) throws IndexerException {
        final int countIndexedItems = ctx.getItems().size();
        updateSolrIndexerCronJobStatistics(ctx, countIndexedItems, 0);
    }

    protected void updateSolrIndexerCronJobStatistics(final IndexerBatchContext ctx, final int additionalIndexedItems, final int additionalErrors) {
        final long indexOperationId = ctx.getIndexOperationId();
        final SolrIndexerCronJobStatistics stats = solrIndexerCronJobStatistics.get(indexOperationId);
        if (stats != null) {
            stats.addIndexedItems(additionalIndexedItems);
            stats.addErrors(additionalErrors);
            updateSolrIndexerCronJobHistory(stats, false);
        }
    }

    protected void updateSolrIndexerCronJobHistory(final SolrIndexerCronJobStatistics stats, final boolean force) {
        final CronJobHistoryModel cronJobHistory = stats.getCronJob().getActiveCronJobHistory();
        if (cronJobHistory != null) {
            if (force || needsUpdate(cronJobHistory)) {
                cronJobHistory.setModifiedtime(new Date());
                final String statusLine = generateStatusLine(stats);
                if (LOG.isInfoEnabled()) {
                    LOG.info("(" + stats.getCronJob().getCode() + ") " + statusLine);
                }
                cronJobHistory.setStatusLine(statusLine);
                getModelService().save(cronJobHistory);
            }
        }
    }

    protected String generateStatusLine(final SolrIndexerCronJobStatistics stats) {
        final long totalIndexedItems = stats.getTotalIndexedItems();
        final int totalIndexErrors = stats.getTotalErrors();
        final long totalIndexDuration =
                (stats.getIndexEndTime() == null ? System.currentTimeMillis() : stats.getIndexEndTime().getTime())
                            - (stats.getIndexStartTime() == null ? stats.getCronJob().getStartTime() : stats.getIndexStartTime()).getTime();
        final NumberFormat nf = NumberFormat.getIntegerInstance(Locale.ENGLISH);
        return String.format("%s item(s) indexed within %s (%s items/s), %s error(s)",
                nf.format(totalIndexedItems),
                totalIndexDuration > 0 ? Duration.ofMillis(totalIndexDuration - (totalIndexDuration % 1000)).toString().substring(2).toLowerCase(Locale.ROOT) : "-",
                nf.format((double) totalIndexedItems / (double)(totalIndexDuration / 1000)), nf.format(totalIndexErrors));
    }

    protected boolean needsUpdate(final CronJobHistoryModel cronJobHistory) {
        return cronJobHistory.getModifiedtime().before(new Date(System.currentTimeMillis() - getUpdateInterval()));
    }

    @Override
    public void afterBatchError(final IndexerBatchContext ctx) throws IndexerException {
        updateSolrIndexerCronJobStatistics(ctx, 0, 1);
    }

    @Override
    public void beforeIndex(final IndexerContext ctx) throws IndexerException {
        hookSolrIndexerCronJob(ctx);
    }

    @Override
    public void afterIndex(final IndexerContext ctx) throws IndexerException {
        unhookSolrIndexerCronJob(ctx);
    }

    @Override
    public void afterIndexError(final IndexerContext ctx) throws IndexerException {
        unhookSolrIndexerCronJob(ctx);
    }

    protected void hookSolrIndexerCronJob(final IndexerContext ctx) {
        final Object currentCronJob = getSessionService().getAttribute("currentCronJob");
        if (currentCronJob instanceof SolrIndexerCronJobModel) {
            final long indexOperationId = ctx.getIndexOperationId();
            solrIndexerCronJobStatistics.put(indexOperationId, new SolrIndexerCronJobStatistics((SolrIndexerCronJobModel) currentCronJob));
        }
    }

    protected void unhookSolrIndexerCronJob(final IndexerContext ctx) {
        final Date indexEndTime = new Date();
        final long indexOperationId = ctx.getIndexOperationId();
        final SolrIndexerCronJobStatistics stats = solrIndexerCronJobStatistics.remove(indexOperationId);
        if (stats != null) {
            stats.setIndexEndTime(indexEndTime);
            updateSolrIndexerCronJobHistory(stats, true);
        }
    }

    protected ModelService getModelService() {
        return modelService;
    }

    @Resource
    public void setModelService(ModelService modelService) {
        this.modelService = modelService;
    }

    protected SessionService getSessionService() {
        return sessionService;
    }

    @Resource
    public void setSessionService(SessionService sessionService) {
        this.sessionService = sessionService;
    }

    public long getUpdateInterval() {
        return updateInterval;
    }

    public void setUpdateInterval(long updateInterval) {
        this.updateInterval = updateInterval;
    }
}

Finally, the listener should be registered, which is achieved with the following Spring configuration.

<bean id="solrIndexerCronJobListener" class="com.discounttire.core.search.solrfacetsearch.indexer.listeners.SolrIndexerCronJobListener">
  <property name="updateInterval" value="#{configurationService.configuration.getInt('solr.indexer.cronjob.status.updateInterval', 60000)}"/>
</bean>

<bean id="solrIndexerCronJobListenerDefinition" parent="solrListenerDefinition">
  <property name="priority" value="2000"/>
  <property name="listener" ref="solrIndexerCronJobListener"/>
</bean>

Limitations

First, this approach works only when the indexing is performed on one node only. It’s the case today by default in SAP Commerce but it could change in the future. Secondly, you might see multiple log messages one after the other published about the same time. It’s normal and due to the fact that updates to CronJobHistory are not synchronized to not slow down operations. Even though modifiedTime is updated manually, it can still happen that batches end at the same time and detect at the same time that CronJobHistory should be updated.

Conclusion

Tracking the speed of the Solr indexing job allowed me to discover, that my indexing job was not scaling because the indexing speed decreased over time. Once the reasons were identified and addressed, it helped to ensure the indexing job was indexing at constant speed. Then, it helped comparing more easily indexing speeds between environments and address infrastructure issues. Finally, it helped detecting early indexing performance issues in pre-production and production environments by comparing the actual indexing speed to the expected one.

Last but not least, if you wonder why I do not set the progress attribute in the active CronJobHistory item, it’s simply because the total count of items to be indexed is not easily available. The context objects passed with the event do not expose this information and customizing the indexing framework to get this information is not simple. If someone has an idea, feel free to share here or to post your own solution referencing this one.