D-Net

import java.io.ByteArrayOutputStream;

import java.io.IOException;

import java.nio.charset.StandardCharsets;

import java.util.List;

import java.util.zip.GZIPOutputStream;

import com.google.common.collect.Lists;

import org.apache.commons.codec.binary.Base64;

import org.apache.commons.lang.exception.ExceptionUtils;

import org.apache.hadoop.conf.Configuration;

import org.apache.solr.client.solrj.impl.CloudSolrServer;

import org.apache.solr.client.solrj.response.SolrPingResponse;

import org.apache.solr.client.solrj.response.UpdateResponse;

	private static final Log log = LogFactory.getLog(IndexFeedMapper.class); // NOPMD by marko on 11/24/08 5:02 PM

	public static final String DNET_RESULT = "dnetResult";

	private InputDocumentFactory documentFactory;

	private CloudSolrServer solrServer;

	private String version;

	private String dsId;

	private int shutdownWaitTime = 10000;

	private int bufferFlushThreshold = 100;

	private ApplyXslt dmfToRecord;

	private List<SolrInputDocument> buffer;

	private int backoffTimeMs = 5000;

	private boolean simulation = false;

	private final static int MAX_INIT_RETRIES = 10;

	private boolean compress = false;

	@Override

		logConfiguration(context.getConfiguration());

		dsId = context.getConfiguration().get(JobParams.INDEX_DSID);

		compress = context.getConfiguration().getBoolean(JobParams.INDEX_FEED_COMPRESS_RESULT, false);

		final String xslt = new String(Base64.decodeBase64(context.getConfiguration().get(JobParams.INDEX_XSLT)));

		log.info("got xslt: \n" + xslt);

		log.info("got version: " + version);

		dmfToRecord = new ApplyXslt(xslt);

		final String baseURL = context.getConfiguration().get(JobParams.INDEX_SOLR_URL);

		log.info("solr server baseURL: " + baseURL);

		final String collection = context.getConfiguration().get(JobParams.INDEX_SOLR_COLLECTION);

		log.info("solr server collection: " + collection);

		int count = 0;

			try {

				count++;

				log.info("initializing solr server...");

				solrServer = new CloudSolrServer(baseURL);

				solrServer.connect();

				solrServer.setParallelUpdates(true);

				final SolrPingResponse rsp = solrServer.ping();

				if (rsp.getStatus() != 0) throw new SolrServerException("bad init status: " + rsp.getStatus());

			} catch (final Throwable e) {

				if (solrServer != null) {

					solrServer.shutdown();

				}

				context.getCounter("index init", e.getMessage()).increment(1);

				Thread.sleep(backoffTimeMs);

	}

	@Override

		String indexRecord = "";

		try {

			doc = documentFactory.parseDocument(version, indexRecord, dsId, DNET_RESULT);

			if ((doc == null) || doc.isEmpty()) throw new EmptySolrDocumentException();

		} catch (final Throwable e) {

			context.getCounter("index feed", "skipped records").increment(1);

			handleError(key, value, context, indexRecord, doc, e);

			return;

		}

		int count = 0;

			try {

				log.info(String.format("failed to feed documents, waiting %dms", backoffTimeMs));

				Thread.sleep(backoffTimeMs);

	public byte[] zip(final String s) {

		if ((s == null) || (s.length() == 0)) {

	private void addDocument(final Context context, final SolrInputDocument doc) throws SolrServerException, IOException, EmptySolrDocumentException {

		}

	private void doAdd(final List<SolrInputDocument> buffer, final Context context) throws SolrServerException, IOException {

			final UpdateResponse rsp = solrServer.add(buffer);

			final long stop = System.currentTimeMillis() - start;

			log.info("feed time for " + buffer.size() + " records : " + HumanTime.exactly(stop) + "\n");

			final int status = rsp.getStatus();

			context.getCounter("index feed", "status code: " + status).increment(buffer.size());

			if (status != 0) throw new SolrServerException("bad status: " + status);

		}

	@Override

			log.info("\nwaiting " + shutdownWaitTime + "ms before shutdown");

			Thread.sleep(shutdownWaitTime);

			solrServer.shutdown();

		} catch (final SolrServerException e) {

			log.error("couldn't shutdown server " + e.getMessage());

		}

	private void handleError(final Text key, final Text value, final Context context, final String indexRecord, final SolrInputDocument doc, final Throwable e)

		context.getCounter("index feed", e.getClass().getName()).increment(1);

		context.write(key, printRottenRecord(context.getTaskAttemptID().toString(), value, indexRecord, doc));

		// e.printStackTrace(System.err);

	}

	private Text printRottenRecord(final String taskid, final Text value, final String indexRecord, final SolrInputDocument doc) {

	private String check(final String label, final Object value) {

	private void logConfiguration(final Configuration conf) {

		log.info("job configutation #################");

		for (final Entry<String, String> e : conf) {

			log.info("'" + e.getKey() + "' : '" + e.getValue() + "'");

		}

		log.info("end of job configutation #################\n\n");

	}

	private String getEntityType(final SolrInputDocument doc) {

		if (!doc.containsKey("oaftype")) return "unknown";

			if (!doc.containsKey("resulttypeid")) return "result";

			return doc.getFieldValue("resulttypeid").toString();

}