core/org.eclipse.smila.solr/code/src/org/eclipse/smila/solr/index/SolrIndexPipelet.java - smila/org.eclipse.smila.core - Git at Google

 /***********************************************************************************************************************
  * Copyright (c) 2008 empolis GmbH and brox IT Solutions GmbH. All rights reserved. This program and the accompanying
  * materials are made available under the terms of the Eclipse Public License v1.0 which accompanies this distribution,
  * and is available at http://www.eclipse.org/legal/epl-v10.html
  *
  * Contributors: Peter Wissel (brox IT Solutions GmbH) - initial API and implementation
  **********************************************************************************************************************/

 package org.eclipse.smila.solr.index;

 import static java.lang.String.format;
 import static org.apache.commons.lang.StringUtils.defaultString;
 import static org.apache.commons.lang.StringUtils.isBlank;
 import static org.apache.commons.lang.StringUtils.isNotBlank;
 import static org.apache.commons.lang.StringUtils.trimToNull;

 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.text.MessageFormat;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;

 import org.apache.commons.lang.NotImplementedException;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.solr.client.solrj.SolrServer;
 import org.apache.solr.client.solrj.SolrServerException;
 import org.apache.solr.client.solrj.response.UpdateResponse;
 import org.apache.solr.common.SolrInputDocument;
 import org.eclipse.smila.blackboard.Blackboard;
 import org.eclipse.smila.blackboard.BlackboardAccessException;
 import org.eclipse.smila.datamodel.Any;
 import org.eclipse.smila.datamodel.AnyMap;
 import org.eclipse.smila.datamodel.AnySeq;
 import org.eclipse.smila.datamodel.Record;
 import org.eclipse.smila.datamodel.xml.XmlSerializationUtils;
 import org.eclipse.smila.processing.Pipelet;
 import org.eclipse.smila.processing.ProcessingException;
 import org.eclipse.smila.processing.parameters.MissingParameterException;
 import org.eclipse.smila.solr.Activator;
 import org.eclipse.smila.solr.SolrConstants;
 import org.eclipse.smila.solr.SolrConstants.AttributeOrAttachment;
 import org.eclipse.smila.solr.SolrConstants.ExecutionMode;
 import org.eclipse.smila.solr.SolrManager;
 import org.eclipse.smila.solr.util.SolrQueryUtils;

 /**
  * SolrIndexPipelet class.
  *
  * @author pwissel
  *
  */
 public class SolrIndexPipelet implements Pipelet {

   /**
    * The configuration error text.
    */
   private static final String CONF_ERROR = "Invalid Pipelet configuration -> ";

   /**
    * The configuration parameter missing error text.
    */
   private static final String CONF_ERROR_PARAMETER_MISSING = CONF_ERROR + "Parameter missing: ";

   /**
    * The log.
    */
   private final Log _log = LogFactory.getLog(SolrIndexPipelet.class);

   /**
    * The configuration map.
    */
   private AnyMap _configuration;

   /**
    * The core fields sequence.
    */
   private AnySeq _coreFieldsSeq;

   /**
    * The execution mode.
    */
   private ExecutionMode _executionMode;

   /**
    * The solr server instance that corresponds to {@link #_defaultCoreName}, attempted to be assigned on each execution
    * of the pipelet.
    */
   private SolrServer _defaultTargetCore;

   /**
    * Optional, if given then it must be alive, otherwise a warning is logged each pipelet execution and processing will
    * continue. It is not an error because it still might be overridden by a dynamic core name.
    */
   private String _defaultCoreName;

   /**
    * {@inheritDoc}
    *
    * @see org.eclipse.smila.processing.IPipelet#configure(org.eclipse.smila.processing.configuration.PipeletConfiguration)
    */
   @Override
   public void configure(AnyMap configuration) throws ProcessingException {
     if (_log.isDebugEnabled()) {
       _log.debug("Configure SolrIndexPipelet");
     }

     _configuration = configuration;
     try {
       if (StringUtils.isBlank(configuration.getStringValue(SolrConstants.EXECUTION_MODE))) {
         throw new MissingParameterException(CONF_ERROR_PARAMETER_MISSING + SolrConstants.EXECUTION_MODE);
       }
       _executionMode = ExecutionMode.valueOf(_configuration.getStringValue(SolrConstants.EXECUTION_MODE));

       switch (_executionMode) {
         case DELETE:
           break;
         case ADD:
         case UPDATE:
           // nothing much happens here yet, see the TODOs in addRecords()
           _coreFieldsSeq = _configuration.getSeq(SolrConstants.CORE_FIELDS);
           for (Any coreFieldMap : _coreFieldsSeq) {
             if (!coreFieldMap.isMap()) {
               throw new ProcessingException("all items must be of type AnyMap in pipelet config: "
                 + SolrConstants.CORE_FIELDS);
             }
           }
           break;
         default:
           throw new NotImplementedException("_executionMode: " + _executionMode);
       }

       // this does nothing except to issue a one time warning about the missing config.
       _defaultCoreName = trimToNull(_configuration.getStringValue(SolrConstants.CORE_NAME));
       if (isBlank(_defaultCoreName)) {
         _log.warn("there is no default core configured in the pipelet config! "
           + "Target core must be set now on each record dynamically via: " + SolrConstants.DYNAMIC_TARGET_CORE);
       }

     } catch (Exception exception) {
       throw new ProcessingException("Error while configure SolrIndexPipelet", exception);
     }

   }

   /**
    * {@inheritDoc}
    *
    * @see org.eclipse.smila.processing.Pipelet#process(org.eclipse.smila.blackboard.Blackboard,
    *      org.eclipse.smila.datamodel.id.String[])
    */
   @Override
   public String[] process(Blackboard blackboard, String[] recordIds) throws ProcessingException {
     if (recordIds != null) {
       if (_defaultCoreName != null) {
         try {
           _defaultTargetCore = getSolrManager().getSolrServer(_defaultCoreName);
         } catch (Exception e) {
           _log
             .warn(format(
               "default core not reachable: %s. Records must have a dynamic core specified or their processing will fail. Exception: %s",
               _defaultCoreName, e.getMessage()));
         }
       }

       switch (_executionMode) {
         case ADD:
           addRecords(blackboard, recordIds);
           break;
         case DELETE:
           deleteRecords(blackboard, recordIds);
           break;
         default:
           throw new NotImplementedException("executionMode: " + _executionMode);
       }

     }
     return recordIds;
   }

   /**
    * Add to solr.
    *
    * @param blackboard
    *          the blackboard.
    * @param recordIds
    *          the record ids.
    * @param server
    * @throws ProcessingException
    *           ProcessingException.
    */
   private void addRecords(Blackboard blackboard, String[] recordIds) throws ProcessingException {
     // since the target core can be set dynamically we must collect all docs per core and this is what this map does
     final Map<SolrServer, Collection<SolrInputDocument>> coreToDocsMap =
       new HashMap<SolrServer, Collection<SolrInputDocument>>();

     final int recordCount = recordIds.length;
     for (String id : recordIds) {
       traceRecord(blackboard, id);

       try {
         final AnyMap metadata = blackboard.getMetadata(id);
         final String dynamicCoreValue = metadata.getStringValue(SolrConstants.DYNAMIC_TARGET_CORE);
         final Collection<SolrInputDocument> solrDocs =
           getInputDocumentsForCore(coreToDocsMap, recordCount, dynamicCoreValue);
         final SolrInputDocument doc = createInputDocument(blackboard, id, metadata);
         solrDocs.add(doc);
         if (_log.isInfoEnabled()) {
           _log.info(format("record added to unsent in-memory document collection.  id: %s core: %s", id,
             defaultString(dynamicCoreValue, _defaultCoreName)));
           if (_log.isTraceEnabled()) {
             _log.trace("solr document: " + doc.toString());
           }
         }
       } catch (Exception e) {
         final String msg = "Error while adding record with id: " + id;
         throw new ProcessingException(msg, e);
       }
     }

     try {
       for (Entry<SolrServer, Collection<SolrInputDocument>> entry : coreToDocsMap.entrySet()) {
         final SolrServer server = entry.getKey();
         final Collection<SolrInputDocument> docs = entry.getValue();
         addDocumentsToServer(server, docs);
       }
     } catch (Exception e) {
       throw new ProcessingException("Error while adding document collection to solr server.", e);
     }
   }

   /**
    * get or create the collection of {@link SolrInputDocument}s to add the record to.
    *
    * @param dynamicCoreValue
    *          may be null
    */
   private Collection<SolrInputDocument> getInputDocumentsForCore(
     final Map<SolrServer, Collection<SolrInputDocument>> coreToDocsMap, int recordCount, String dynamicCoreValue)
     throws SolrServerException, IOException, ProcessingException {

     final SolrServer targetCore = getTargetCore(dynamicCoreValue);
     Collection<SolrInputDocument> solrDocs = coreToDocsMap.get(targetCore);
     if (solrDocs == null) {
       solrDocs = new ArrayList<SolrInputDocument>(recordCount);
       coreToDocsMap.put(targetCore, solrDocs);
     }
     return solrDocs;
   }

   /**
    * get the target Solr core for the given record.
    *
    * @param dynamicCoreValue
    *          TODO
    */
   private SolrServer getTargetCore(String dynamicCoreValue) throws SolrServerException, IOException,
     ProcessingException {

     if (isNotBlank(dynamicCoreValue)) {
       return getSolrManager().getSolrServer(dynamicCoreValue);
     } else {
       if (_defaultTargetCore == null) {
         throw new ProcessingException(
           "no dynamic core in record given while default core is null. check if the pipelet defines a valid default core or that the record carries the dynamicCore attribute");
       }
       return _defaultTargetCore;
     }
   }

   /** create a {@link SolrInputDocument} from the given record using the specified mapping. */
   private SolrInputDocument createInputDocument(Blackboard blackboard, String id, final AnyMap metadata)
     throws ProcessingException, BlackboardAccessException, UnsupportedEncodingException {
     final SolrInputDocument doc = new SolrInputDocument();
     // add id field
     doc.addField(SolrConstants.CORE_FIELD_ID, id);
     // set optional boost factor
     final Double boostFactor = metadata.getDoubleValue(SolrConstants.DOC_BOOST);
     if (boostFactor != null) {
       doc.setDocumentBoost(boostFactor.floatValue());
     }
     /*
      * PERF: move these checks into config() | TM @ Jun 7, 2011
      *
      * PERF: transform the map structure into a faster structure and do conversion and setting of default values there |
      * TM @ Jun 7, 2011
      *
      * BETTER: check that no target core field is mapped twice (from diff. sourc fields), as it will be overwritten.
      * since there are rare use cases for this (e.g. processing ensures that just on of the source field is actually
      * set), we just should issue a warning here) | TM @ Jun 7, 2011
      */
     for (Any coreFieldMap : _coreFieldsSeq) {
       final AnyMap fieldMap = coreFieldMap.asMap(); // configure validates this in advance
       final String fieldName = getFieldName(fieldMap);
       final String recSourceName = getRecordSourceName(fieldMap, fieldName);
       final AttributeOrAttachment recSourceType = getRecordSourceType(fieldMap, fieldName);
       /* TODO | TM | add boost on field level | TM @ May 20, 2011 */
       switch (recSourceType) {
         case ATTRIBUTE:
           createFieldFromAttribute(doc, fieldName, metadata, recSourceName);
           break;
         case ATTACHMENT:
           createFieldFromAttachment(doc, fieldName, blackboard, id, recSourceName);
           break;
         default:
           throw new NotImplementedException("recSourceType: " + recSourceType);
       }
     }
     return doc;
   }

   /** get the field name from the record-document-mapping. */
   private String getFieldName(final AnyMap fieldMap) throws MissingParameterException {
     final String fieldName = fieldMap.getStringValue(SolrConstants.CORE_FIELD_NAME);
     if (StringUtils.isBlank(fieldName)) {
       throw new MissingParameterException(CONF_ERROR_PARAMETER_MISSING + SolrConstants.CORE_FIELD_NAME);
     }
     return fieldName;
   }

   /** get the attribute or attachment name from the record-document-mapping. */
   private String getRecordSourceName(final AnyMap fieldMap, final String fieldName) {
     String recSourceName = fieldMap.getStringValue(SolrConstants.SOURCE_NAME);
     if (StringUtils.isBlank(recSourceName)) {
       recSourceName = fieldName;
       if (_log.isTraceEnabled()) {
         _log.trace(format("core field mapping %s: no %s config'ed. Defaulting to field name", fieldName,
           SolrConstants.SOURCE_NAME));
       }
     }
     return recSourceName;
   }

   /** get the record source type from record-document-mapping. */
   private AttributeOrAttachment getRecordSourceType(final AnyMap fieldMap, final String fieldName) {
     final String recSourceTypeString = fieldMap.getStringValue(SolrConstants.SOURCE_TYPE);
     final AttributeOrAttachment recSourceType;
     if (isBlank(recSourceTypeString)) {
       recSourceType = AttributeOrAttachment.ATTRIBUTE;
       if (_log.isTraceEnabled()) {
         _log.trace(format("core field mapping %s: no %s config'ed. Defaulting to %s", fieldName,
           SolrConstants.SOURCE_TYPE, recSourceType));
       }
     } else {
       recSourceType = AttributeOrAttachment.valueOf(recSourceTypeString.toUpperCase());
     }
     return recSourceType;
   }

   /** create a Solr field from a record attribute value. */
   private void createFieldFromAttribute(final SolrInputDocument doc, final String fieldName, final AnyMap metadata,
     String recordAttribute) throws ProcessingException {
     final Any any = metadata.get(recordAttribute);
     if (any != null) {
       if (any.isValue()) {
         doc.addField(fieldName, any.asValue().getObject());
       } else if (any.isSeq()) {
         final AnySeq asSeq = any.asSeq();
         for (Any seqValues : asSeq) {
           doc.addField(fieldName, seqValues.asValue().getObject());
         }
       } else {
         throw new ProcessingException(
           "value type for indexing in solr not supported. Must be one of Value, Seq but is: " + any.getValueType());
       }
     } else {
       if (_log.isTraceEnabled()) {// this case is quite common and hence only logged @ trace
         _log.trace("Record doesn't contain an attribute named: " + recordAttribute);
       }
     }
   }

   /** create a Solr field from a record attachment. */
   private void createFieldFromAttachment(final SolrInputDocument doc, final String fieldName,
     Blackboard blackboard, String id, String attachmentName) throws BlackboardAccessException,
     UnsupportedEncodingException {
     final byte[] value = blackboard.getAttachmentAsBytes(id, attachmentName);
     if (value != null) {
       final String string = new String(value, "UTF-8");
       doc.addField(fieldName, string);
     } else {
       // this case is quite common and hence only logged @ trace
       if (_log.isTraceEnabled() && !blackboard.hasAttachment(id, attachmentName)) {
         _log.trace("Record doesn't have an attachment named: " + attachmentName);
       }
     }
   }

   /** send the created {@link SolrInputDocument}s to a server. */
   private void addDocumentsToServer(final SolrServer server, final Collection<SolrInputDocument> docs)
     throws SolrServerException, IOException, ProcessingException {
     final UpdateResponse solrResponse = server.add(docs);
     if (SolrQueryUtils.responseStatusIsError(solrResponse)) {
       throw new ProcessingException("Error reported by solr reponse with status: " + solrResponse.getStatus());
     }
     // TODO: What is best way to commit?
     // _defaultTargetCore.commit(); // use autoCommit in solrconfig.xml
     if (_log.isDebugEnabled()) {
       _log.info(MessageFormat.format("document collection was added to solr in {1} ms, doc count = {0} ", //
         docs.size(), solrResponse.getElapsedTime()));
     }
   }

   /** Log serialized record on trace level. */
   private void traceRecord(Blackboard blackboard, String id) {
     if (_log.isTraceEnabled()) {
       try {
         final Record record = blackboard.getRecord(id);
         final String serialize2string = XmlSerializationUtils.serialize2string(record).replace('\n', ' ');
         _log.trace("processing record: " + serialize2string);
       } catch (BlackboardAccessException e) {
         _log.trace("error on serializing the record for logging: ", e);
       }
     }
   }

   /** Delete from solr. */
   private void deleteRecords(Blackboard blackboard, String[] recordIds) throws ProcessingException {
     for (String id : recordIds) {
       try {
         final AnyMap metadata = blackboard.getMetadata(id);
         final String indexName =
           StringUtils.defaultIfEmpty(metadata.getStringValue(SolrConstants.DYNAMIC_TARGET_CORE), _defaultCoreName);
         final SolrServer server = getSolrManager().getSolrServer(indexName);

         /*
          * NOTE: solr offers to delete a set of ids. but we deliberately don't do this for logging purposes | tmenzel @
          * May 20, 2011
          */
         final UpdateResponse deleteResponse = server.deleteById(id);
         if (SolrQueryUtils.responseStatusIsError(deleteResponse)) {
           throw new ProcessingException("Error reported by solr reponse while delete record with id: " + id);
         }
         if (_log.isDebugEnabled()) {
           final String msg =
             MessageFormat.format("Record deleted: Id: {0} Index: {1} time: {2}ms.", id,
               deleteResponse.getRequestUrl(), deleteResponse.getElapsedTime());
           _log.debug(msg);
         }
       } catch (Exception e) {
         throw new ProcessingException("Error while delete record with id: " + id, e);
       }
     }
   }

   /**
    * Get SolrManager.
    *
    * @return the SolrManager.
    */
   public SolrManager getSolrManager() {
     /*
      * WORKAROUND: must be here instead of ctor/configure() due to init problems. suspicion: pipelet tracker calling
      * configure() before the activator is called, despite it's lazy setting | TM @ Jun 7, 2011
      */

     return Activator.getInstance().getSolrManager();
   }

 }
	/***********************************************************************************************************************
	* Copyright (c) 2008 empolis GmbH and brox IT Solutions GmbH. All rights reserved. This program and the accompanying
	* materials are made available under the terms of the Eclipse Public License v1.0 which accompanies this distribution,
	* and is available at http://www.eclipse.org/legal/epl-v10.html
	*
	* Contributors: Peter Wissel (brox IT Solutions GmbH) - initial API and implementation
	**********************************************************************************************************************/

	package org.eclipse.smila.solr.index;

	import static java.lang.String.format;
	import static org.apache.commons.lang.StringUtils.defaultString;
	import static org.apache.commons.lang.StringUtils.isBlank;
	import static org.apache.commons.lang.StringUtils.isNotBlank;
	import static org.apache.commons.lang.StringUtils.trimToNull;

	import java.io.IOException;
	import java.io.UnsupportedEncodingException;
	import java.text.MessageFormat;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.HashMap;
	import java.util.Map;
	import java.util.Map.Entry;

	import org.apache.commons.lang.NotImplementedException;
	import org.apache.commons.lang.StringUtils;
	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;
	import org.apache.solr.client.solrj.SolrServer;
	import org.apache.solr.client.solrj.SolrServerException;
	import org.apache.solr.client.solrj.response.UpdateResponse;
	import org.apache.solr.common.SolrInputDocument;
	import org.eclipse.smila.blackboard.Blackboard;
	import org.eclipse.smila.blackboard.BlackboardAccessException;
	import org.eclipse.smila.datamodel.Any;
	import org.eclipse.smila.datamodel.AnyMap;
	import org.eclipse.smila.datamodel.AnySeq;
	import org.eclipse.smila.datamodel.Record;
	import org.eclipse.smila.datamodel.xml.XmlSerializationUtils;
	import org.eclipse.smila.processing.Pipelet;
	import org.eclipse.smila.processing.ProcessingException;
	import org.eclipse.smila.processing.parameters.MissingParameterException;
	import org.eclipse.smila.solr.Activator;
	import org.eclipse.smila.solr.SolrConstants;
	import org.eclipse.smila.solr.SolrConstants.AttributeOrAttachment;
	import org.eclipse.smila.solr.SolrConstants.ExecutionMode;
	import org.eclipse.smila.solr.SolrManager;
	import org.eclipse.smila.solr.util.SolrQueryUtils;

	/**
	* SolrIndexPipelet class.
	*
	* @author pwissel
	*
	*/
	public class SolrIndexPipelet implements Pipelet {

	/**
	* The configuration error text.
	*/
	private static final String CONF_ERROR = "Invalid Pipelet configuration -> ";

	/**
	* The configuration parameter missing error text.
	*/
	private static final String CONF_ERROR_PARAMETER_MISSING = CONF_ERROR + "Parameter missing: ";

	/**
	* The log.
	*/
	private final Log _log = LogFactory.getLog(SolrIndexPipelet.class);

	/**
	* The configuration map.
	*/
	private AnyMap _configuration;

	/**
	* The core fields sequence.
	*/
	private AnySeq _coreFieldsSeq;

	/**
	* The execution mode.
	*/
	private ExecutionMode _executionMode;

	/**
	* The solr server instance that corresponds to {@link #_defaultCoreName}, attempted to be assigned on each execution
	* of the pipelet.
	*/
	private SolrServer _defaultTargetCore;

	/**
	* Optional, if given then it must be alive, otherwise a warning is logged each pipelet execution and processing will
	* continue. It is not an error because it still might be overridden by a dynamic core name.
	*/
	private String _defaultCoreName;

	/**
	* {@inheritDoc}
	*
	* @see org.eclipse.smila.processing.IPipelet#configure(org.eclipse.smila.processing.configuration.PipeletConfiguration)
	*/
	@Override
	public void configure(AnyMap configuration) throws ProcessingException {
	if (_log.isDebugEnabled()) {
	_log.debug("Configure SolrIndexPipelet");
	}

	_configuration = configuration;
	try {
	if (StringUtils.isBlank(configuration.getStringValue(SolrConstants.EXECUTION_MODE))) {
	throw new MissingParameterException(CONF_ERROR_PARAMETER_MISSING + SolrConstants.EXECUTION_MODE);
	}
	_executionMode = ExecutionMode.valueOf(_configuration.getStringValue(SolrConstants.EXECUTION_MODE));

	switch (_executionMode) {
	case DELETE:
	break;
	case ADD:
	case UPDATE:
	// nothing much happens here yet, see the TODOs in addRecords()
	_coreFieldsSeq = _configuration.getSeq(SolrConstants.CORE_FIELDS);
	for (Any coreFieldMap : _coreFieldsSeq) {
	if (!coreFieldMap.isMap()) {
	throw new ProcessingException("all items must be of type AnyMap in pipelet config: "
	+ SolrConstants.CORE_FIELDS);
	}
	}
	break;
	default:
	throw new NotImplementedException("_executionMode: " + _executionMode);
	}

	// this does nothing except to issue a one time warning about the missing config.
	_defaultCoreName = trimToNull(_configuration.getStringValue(SolrConstants.CORE_NAME));
	if (isBlank(_defaultCoreName)) {
	_log.warn("there is no default core configured in the pipelet config! "
	+ "Target core must be set now on each record dynamically via: " + SolrConstants.DYNAMIC_TARGET_CORE);
	}

	} catch (Exception exception) {
	throw new ProcessingException("Error while configure SolrIndexPipelet", exception);
	}

	}

	/**
	* {@inheritDoc}
	*
	* @see org.eclipse.smila.processing.Pipelet#process(org.eclipse.smila.blackboard.Blackboard,
	* org.eclipse.smila.datamodel.id.String[])
	*/
	@Override
	public String[] process(Blackboard blackboard, String[] recordIds) throws ProcessingException {
	if (recordIds != null) {
	if (_defaultCoreName != null) {
	try {
	_defaultTargetCore = getSolrManager().getSolrServer(_defaultCoreName);
	} catch (Exception e) {
	_log
	.warn(format(
	"default core not reachable: %s. Records must have a dynamic core specified or their processing will fail. Exception: %s",
	_defaultCoreName, e.getMessage()));
	}
	}

	switch (_executionMode) {
	case ADD:
	addRecords(blackboard, recordIds);
	break;
	case DELETE:
	deleteRecords(blackboard, recordIds);
	break;
	default:
	throw new NotImplementedException("executionMode: " + _executionMode);
	}

	}
	return recordIds;
	}

	/**
	* Add to solr.
	*
	* @param blackboard
	* the blackboard.
	* @param recordIds
	* the record ids.
	* @param server
	* @throws ProcessingException
	* ProcessingException.
	*/
	private void addRecords(Blackboard blackboard, String[] recordIds) throws ProcessingException {
	// since the target core can be set dynamically we must collect all docs per core and this is what this map does
	final Map<SolrServer, Collection<SolrInputDocument>> coreToDocsMap =
	new HashMap<SolrServer, Collection<SolrInputDocument>>();

	final int recordCount = recordIds.length;
	for (String id : recordIds) {
	traceRecord(blackboard, id);

	try {
	final AnyMap metadata = blackboard.getMetadata(id);
	final String dynamicCoreValue = metadata.getStringValue(SolrConstants.DYNAMIC_TARGET_CORE);
	final Collection<SolrInputDocument> solrDocs =
	getInputDocumentsForCore(coreToDocsMap, recordCount, dynamicCoreValue);
	final SolrInputDocument doc = createInputDocument(blackboard, id, metadata);
	solrDocs.add(doc);
	if (_log.isInfoEnabled()) {
	_log.info(format("record added to unsent in-memory document collection. id: %s core: %s", id,
	defaultString(dynamicCoreValue, _defaultCoreName)));
	if (_log.isTraceEnabled()) {
	_log.trace("solr document: " + doc.toString());
	}
	}
	} catch (Exception e) {
	final String msg = "Error while adding record with id: " + id;
	throw new ProcessingException(msg, e);
	}
	}

	try {
	for (Entry<SolrServer, Collection<SolrInputDocument>> entry : coreToDocsMap.entrySet()) {
	final SolrServer server = entry.getKey();
	final Collection<SolrInputDocument> docs = entry.getValue();
	addDocumentsToServer(server, docs);
	}
	} catch (Exception e) {
	throw new ProcessingException("Error while adding document collection to solr server.", e);
	}
	}

	/**
	* get or create the collection of {@link SolrInputDocument}s to add the record to.
	*
	* @param dynamicCoreValue
	* may be null
	*/
	private Collection<SolrInputDocument> getInputDocumentsForCore(
	final Map<SolrServer, Collection<SolrInputDocument>> coreToDocsMap, int recordCount, String dynamicCoreValue)
	throws SolrServerException, IOException, ProcessingException {

	final SolrServer targetCore = getTargetCore(dynamicCoreValue);
	Collection<SolrInputDocument> solrDocs = coreToDocsMap.get(targetCore);
	if (solrDocs == null) {
	solrDocs = new ArrayList<SolrInputDocument>(recordCount);
	coreToDocsMap.put(targetCore, solrDocs);
	}
	return solrDocs;
	}

	/**
	* get the target Solr core for the given record.
	*
	* @param dynamicCoreValue
	* TODO
	*/
	private SolrServer getTargetCore(String dynamicCoreValue) throws SolrServerException, IOException,
	ProcessingException {

	if (isNotBlank(dynamicCoreValue)) {
	return getSolrManager().getSolrServer(dynamicCoreValue);
	} else {
	if (_defaultTargetCore == null) {
	throw new ProcessingException(
	"no dynamic core in record given while default core is null. check if the pipelet defines a valid default core or that the record carries the dynamicCore attribute");
	}
	return _defaultTargetCore;
	}
	}

	/** create a {@link SolrInputDocument} from the given record using the specified mapping. */
	private SolrInputDocument createInputDocument(Blackboard blackboard, String id, final AnyMap metadata)
	throws ProcessingException, BlackboardAccessException, UnsupportedEncodingException {
	final SolrInputDocument doc = new SolrInputDocument();
	// add id field
	doc.addField(SolrConstants.CORE_FIELD_ID, id);
	// set optional boost factor
	final Double boostFactor = metadata.getDoubleValue(SolrConstants.DOC_BOOST);
	if (boostFactor != null) {
	doc.setDocumentBoost(boostFactor.floatValue());
	}
	/*
	* PERF: move these checks into config() \| TM @ Jun 7, 2011
	*
	* PERF: transform the map structure into a faster structure and do conversion and setting of default values there \|
	* TM @ Jun 7, 2011
	*
	* BETTER: check that no target core field is mapped twice (from diff. sourc fields), as it will be overwritten.
	* since there are rare use cases for this (e.g. processing ensures that just on of the source field is actually
	* set), we just should issue a warning here) \| TM @ Jun 7, 2011
	*/
	for (Any coreFieldMap : _coreFieldsSeq) {
	final AnyMap fieldMap = coreFieldMap.asMap(); // configure validates this in advance
	final String fieldName = getFieldName(fieldMap);
	final String recSourceName = getRecordSourceName(fieldMap, fieldName);
	final AttributeOrAttachment recSourceType = getRecordSourceType(fieldMap, fieldName);
	/* TODO \| TM \| add boost on field level \| TM @ May 20, 2011 */
	switch (recSourceType) {
	case ATTRIBUTE:
	createFieldFromAttribute(doc, fieldName, metadata, recSourceName);
	break;
	case ATTACHMENT:
	createFieldFromAttachment(doc, fieldName, blackboard, id, recSourceName);
	break;
	default:
	throw new NotImplementedException("recSourceType: " + recSourceType);
	}
	}
	return doc;
	}

	/** get the field name from the record-document-mapping. */
	private String getFieldName(final AnyMap fieldMap) throws MissingParameterException {
	final String fieldName = fieldMap.getStringValue(SolrConstants.CORE_FIELD_NAME);
	if (StringUtils.isBlank(fieldName)) {
	throw new MissingParameterException(CONF_ERROR_PARAMETER_MISSING + SolrConstants.CORE_FIELD_NAME);
	}
	return fieldName;
	}

	/** get the attribute or attachment name from the record-document-mapping. */
	private String getRecordSourceName(final AnyMap fieldMap, final String fieldName) {
	String recSourceName = fieldMap.getStringValue(SolrConstants.SOURCE_NAME);
	if (StringUtils.isBlank(recSourceName)) {
	recSourceName = fieldName;
	if (_log.isTraceEnabled()) {
	_log.trace(format("core field mapping %s: no %s config'ed. Defaulting to field name", fieldName,
	SolrConstants.SOURCE_NAME));
	}
	}
	return recSourceName;
	}

	/** get the record source type from record-document-mapping. */
	private AttributeOrAttachment getRecordSourceType(final AnyMap fieldMap, final String fieldName) {
	final String recSourceTypeString = fieldMap.getStringValue(SolrConstants.SOURCE_TYPE);
	final AttributeOrAttachment recSourceType;
	if (isBlank(recSourceTypeString)) {
	recSourceType = AttributeOrAttachment.ATTRIBUTE;
	if (_log.isTraceEnabled()) {
	_log.trace(format("core field mapping %s: no %s config'ed. Defaulting to %s", fieldName,
	SolrConstants.SOURCE_TYPE, recSourceType));
	}
	} else {
	recSourceType = AttributeOrAttachment.valueOf(recSourceTypeString.toUpperCase());
	}
	return recSourceType;
	}

	/** create a Solr field from a record attribute value. */
	private void createFieldFromAttribute(final SolrInputDocument doc, final String fieldName, final AnyMap metadata,
	String recordAttribute) throws ProcessingException {
	final Any any = metadata.get(recordAttribute);
	if (any != null) {
	if (any.isValue()) {
	doc.addField(fieldName, any.asValue().getObject());
	} else if (any.isSeq()) {
	final AnySeq asSeq = any.asSeq();
	for (Any seqValues : asSeq) {
	doc.addField(fieldName, seqValues.asValue().getObject());
	}
	} else {
	throw new ProcessingException(
	"value type for indexing in solr not supported. Must be one of Value, Seq but is: " + any.getValueType());
	}
	} else {
	if (_log.isTraceEnabled()) {// this case is quite common and hence only logged @ trace
	_log.trace("Record doesn't contain an attribute named: " + recordAttribute);
	}
	}
	}

	/** create a Solr field from a record attachment. */
	private void createFieldFromAttachment(final SolrInputDocument doc, final String fieldName,
	Blackboard blackboard, String id, String attachmentName) throws BlackboardAccessException,
	UnsupportedEncodingException {
	final byte[] value = blackboard.getAttachmentAsBytes(id, attachmentName);
	if (value != null) {
	final String string = new String(value, "UTF-8");
	doc.addField(fieldName, string);
	} else {
	// this case is quite common and hence only logged @ trace
	if (_log.isTraceEnabled() && !blackboard.hasAttachment(id, attachmentName)) {
	_log.trace("Record doesn't have an attachment named: " + attachmentName);
	}
	}
	}

	/** send the created {@link SolrInputDocument}s to a server. */
	private void addDocumentsToServer(final SolrServer server, final Collection<SolrInputDocument> docs)
	throws SolrServerException, IOException, ProcessingException {
	final UpdateResponse solrResponse = server.add(docs);
	if (SolrQueryUtils.responseStatusIsError(solrResponse)) {
	throw new ProcessingException("Error reported by solr reponse with status: " + solrResponse.getStatus());
	}
	// TODO: What is best way to commit?
	// _defaultTargetCore.commit(); // use autoCommit in solrconfig.xml
	if (_log.isDebugEnabled()) {
	_log.info(MessageFormat.format("document collection was added to solr in {1} ms, doc count = {0} ", //
	docs.size(), solrResponse.getElapsedTime()));
	}
	}

	/** Log serialized record on trace level. */
	private void traceRecord(Blackboard blackboard, String id) {
	if (_log.isTraceEnabled()) {
	try {
	final Record record = blackboard.getRecord(id);
	final String serialize2string = XmlSerializationUtils.serialize2string(record).replace('\n', ' ');
	_log.trace("processing record: " + serialize2string);
	} catch (BlackboardAccessException e) {
	_log.trace("error on serializing the record for logging: ", e);
	}
	}
	}

	/** Delete from solr. */
	private void deleteRecords(Blackboard blackboard, String[] recordIds) throws ProcessingException {
	for (String id : recordIds) {
	try {
	final AnyMap metadata = blackboard.getMetadata(id);
	final String indexName =
	StringUtils.defaultIfEmpty(metadata.getStringValue(SolrConstants.DYNAMIC_TARGET_CORE), _defaultCoreName);
	final SolrServer server = getSolrManager().getSolrServer(indexName);

	/*
	* NOTE: solr offers to delete a set of ids. but we deliberately don't do this for logging purposes \| tmenzel @
	* May 20, 2011
	*/
	final UpdateResponse deleteResponse = server.deleteById(id);
	if (SolrQueryUtils.responseStatusIsError(deleteResponse)) {
	throw new ProcessingException("Error reported by solr reponse while delete record with id: " + id);
	}
	if (_log.isDebugEnabled()) {
	final String msg =
	MessageFormat.format("Record deleted: Id: {0} Index: {1} time: {2}ms.", id,
	deleteResponse.getRequestUrl(), deleteResponse.getElapsedTime());
	_log.debug(msg);
	}
	} catch (Exception e) {
	throw new ProcessingException("Error while delete record with id: " + id, e);
	}
	}
	}

	/**
	* Get SolrManager.
	*
	* @return the SolrManager.
	*/
	public SolrManager getSolrManager() {
	/*
	* WORKAROUND: must be here instead of ctor/configure() due to init problems. suspicion: pipelet tracker calling
	* configure() before the activator is called, despite it's lazy setting \| TM @ Jun 7, 2011
	*/

	return Activator.getInstance().getSolrManager();
	}

	}