Possible threading problem in SI4T

Question

Is anyone aware of a threading issues within SI4T?

We have a custom Elasticsearch indexer plugged into SI4T that uses JEST to check and create necessary Elasticsearch indexes based on the publication id (e.g. publication_10) passed in to the commit method and then index documents into these indexes. The code is based pretty much on the SOLR example that comes with with SI4T, apart from the Elasticsearch/JEST parts.

Our blueprint is a traditional diamond (content/templates) and has a master website publication for pages that several child publications inherit from.

We've noticed that when we publish pages via the static components that are used on them, from the master (with the include in Child Publications option) we can see a load of transactions in the queue - all as expected apart from we end up with pages from different publications indexed into the wrong index - so index "publication_36" has a document with id "tcm:51-xxxx-64".

I don't quite understand storage extensions or the effect of a multi-threaded Deployer - but it seems like because parts of SI4T are singleton, the multi-threaded Deployer is altering the publication id value passed in to the SI4T storage classes.

UPDATE

cd_storage:

<Storage 
    Type="persistence" Id="si4tdb" dialect="MSSQL" 
    Class="com.tridion.storage.si4t.JPASearchDAOFactory">
    <Pool Type="jdbc" Size="5" MonitorInterval="60" IdleTimeout="120" CheckoutTimeout="120" />
    <DataSource Class="com.microsoft.sqlserver.jdbc.SQLServerDataSource">
        <Property Name="serverName" Value="...." />
        <Property Name="portNumber" Value="1433" />
        <Property Name="databaseName" Value="Tridion_Broker" />
        <Property Name="user" Value="..." />
        <Property Name="password" Value="..." />                        
    </DataSource>
    <Indexer 
        Class="org.si4t.elasticsearch.ElasticsearchIndexer" 
        Url="http://localhost:9200" 
        Clustername="http"
        DocExtensions="pdf,docx,doc,xls,xlsx,pptx,ppt" 
        Mapping=" ..."/>
</Storage>


<Storage Type="filesystem" 
        Class="com.tridion.storage.si4t.FSSearchDAOFactory" 
        Id="301FileSystem" defaultFilesystem="false" defaultStorage="false">                  
    <Root Path="D:\Live\Redirects" />
    <Indexer 
        Class="org.si4t.elasticsearch.ElasticsearchIndexer" 
        Clustername="http" 
        Url="http://localhost:9200"
        DocExtensions="pdf,docx,doc,xls,xlsx,pptx,ppt,txt" 
        Mapping="..."/>
</Storage>

<Storage Type="filesystem" 
        Class="com.tridion.storage.si4t.FSSearchDAOFactory" 
        Id="si4tFile" defaultFilesystem="false" defaultStorage="false">               
    <Root Path="D:\Live\Binaries" />
    <Indexer 
        Class="org.si4t.elasticsearch.ElasticsearchIndexer" 
        Clustername="http" 
        Url="http://localhost:9200"
        DocExtensions="pdf,docx,doc,xls,xlsx,pptx,ppt" 
        Mapping="..."/>
</Storage>

ELasticsearch indexer can be found here. The approach was to use the publication id passed to the commit method to form the index name - check if it exists (create if not) - and then index documents using this index name.

Example log file can be found here. I can see on line 1116 for exmaple, where several pages across publications have all gone into index "publication_36".

pool-1-thread-6 2015-03-06 16:25:42,072 DEBUG ElasticsearchIndexer - {"took":483,"errors":false,"items":[{"index":{"_index":"publication_36","_type":"Pages","_id":"tcm:52-1786-64","_version":1,"status":201}},{"index":{"_index":"publication_36","_type":"Pages","_id":"tcm:51-1786-64","_version":1,"status":201}},{"index":{"_index":"publication_36","_type":"Pages","_id":"tcm:49-1786-64","_version":1,"status":201}},{"index":{"_index":"publication_36","_type":"Pages","_id":"tcm:47-1786-64","_version":1,"status":201}},{"index":{"_index":"publication_36","_type":"Pages","_id":"tcm:40-1786-64","_version":1,"status":201}},{"index":{"_index":"publication_36","_type":"Pages","_id":"tcm:36-1786-64","_version":1,"status":201}}]}

Answer 1

My suspicion is the the problem lies in ElasticSearchIndexer. Consider the following code:

private String publicationId = null;
private String indexname = null;
//.. truncated

public void commit(String publicationId) throws IndexingException 
{

    System.out.println("Commit: "+ publicationId);
    log.debug("ElasticsearchIndexer - commit called: " + publicationId);

    this.publicationId = publicationId;
    this.indexname = "publication_" + this.publicationId;

The DAO factories can call one commit for multiple transactions. This means that say you publish out a page with the option "Also publish in Child Publications" on, then you will have multiple transactions across various publications, in one commit.

Secondly, the way the Indexer works, is by building a stack of items to add, remove or update. What this means is that, you can have items from different publications on for instance the itemAdds stack, which will get indexed in the same index, because you use the this.indexname property for all items on that stack.

Solution

Construct the index name for each individual items for each commit* method, like so:

log.info("Adding pages and component presentations");

for (Map.Entry<String, SearchIndexData> entry : itemAdds.entrySet())
{
    SearchIndexData data = entry.getValue();

    String id = data.getUniqueIndexId();

    // NEW
    String publicationId = data.getPublicationItemId();
    String currentIndexName = "publication_" + publicationId;

    // Truncated

    builder.addAction(new Index.Builder(source).index(currentIndexName).type(this.INDEX_TYPE).id(id).build());
}   

This will ensure each item is going into its proper index.

What is the use of the class variable this.indexname?

The mechanism to send the publication Id with the call to commit is actually meant to be able to configure cd_storage_conf.xml to have a separate index URL / name per publication Id. Since this is not configured in your cd_storage_conf.xml, you get behaviour like this. However, for your case I recommend to fix this in code as described above.