<commit_before>/**
 * Copyright 2016, RadiantBlue Technologies, Inc.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 **/
package ingest.messaging;

import ingest.event.IngestEvent;
import ingest.inspect.Inspector;
import ingest.utility.IngestUtilities;

import java.io.IOException;
import java.util.concurrent.Future;

import messaging.job.JobMessageFactory;
import messaging.job.WorkerCallback;
import model.data.DataResource;
import model.data.FileRepresentation;
import model.data.location.FileLocation;
import model.data.location.FolderShare;
import model.data.location.S3FileStore;
import model.job.Job;
import model.job.JobProgress;
import model.job.result.type.DataResult;
import model.job.result.type.ErrorResult;
import model.job.type.IngestJob;
import model.job.type.SearchMetadataIngestJob;
import model.response.PiazzaResponse;
import model.status.StatusUpdate;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import util.PiazzaLogger;
import util.UUIDFactory;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mongodb.MongoException;

/**
 * Worker class that handles a specific Ingest Job. The threads are managed by
 * the IngestThreadManager class.
 * 
 * @author Patrick.Doody & Sonny.Saniev & Russell.Orf
 * 
 */
@Component
public class IngestWorker {
	@Value("${SPACE}")
	private String SPACE;
	@Value("${workflow.event.id}")
	private String EVENT_ID;
	@Value("${workflow.url}")
	private String WORKFLOW_URL;
	@Value("${search.url}")
	private String SEARCH_URL;
	@Value("${workflow.endpoint}")
	private String WORKFLOW_ENDPOINT;
	@Value("${search.ingest.endpoint}")
	private String SEARCH_ENDPOINT;
	@Value("${vcap.services.pz-blobstore.credentials.bucket}")
	private String AMAZONS3_BUCKET_NAME;

	@Autowired
	private PiazzaLogger logger;
	@Autowired
	private Inspector inspector;
	@Autowired
	private IngestUtilities ingestUtilities;
	@Autowired
	private UUIDFactory uuidFactory;

	/**
	 * Creates a new Worker Thread for the specified Kafka Message containing an
	 * Ingest Job.
	 * 
	 * @param consumerRecord
	 *            The Kafka Message containing the Job.
	 * @param producer
	 *            The Kafka producer, used to send update messages
	 * @param callback
	 *            The callback that will be invoked when this Job has finished
	 *            processing (error or success, regardless)
	 */
	@Async
	public Future<DataResource> run(ConsumerRecord<String, String> consumerRecord, Producer<String, String> producer,
			WorkerCallback callback) {
		DataResource dataResource = null;

		try {
			// Log
			logger.log(String.format("Processing Data Load for Topic %s with Key %s", consumerRecord.topic(),
					consumerRecord.key()), PiazzaLogger.INFO);

			// Parse the Job from the Kafka Message
			ObjectMapper mapper = new ObjectMapper();
			Job job = mapper.readValue(consumerRecord.value(), Job.class);
			IngestJob ingestJob = (IngestJob) job.getJobType();
			// Get the description of the Data to be ingested
			dataResource = ingestJob.getData();

			// Assign a Resource ID to the incoming DataResource.
			if (dataResource.getDataId() == null) {
				String dataId = uuidFactory.getUUID();
				dataResource.setDataId(dataId);
			}

			// Log what we're going to Ingest
			logger.log(String.format(
					"Inspected Load Job; begin Loading Data %s of Type %s. Hosted: %s with Job ID of %s",
					dataResource.getDataId(), dataResource.getDataType().getType(), ingestJob.getHost().toString(),
					job.getJobId()), PiazzaLogger.INFO);

			// Update Status on Handling
			JobProgress jobProgress = new JobProgress(0);
			StatusUpdate statusUpdate = new StatusUpdate(StatusUpdate.STATUS_RUNNING, jobProgress);
			producer.send(JobMessageFactory.getUpdateStatusMessage(consumerRecord.key(), statusUpdate, SPACE));

			if (ingestJob.getData().getDataType() instanceof FileRepresentation) {
				FileRepresentation fileRep = (FileRepresentation) ingestJob.getData().getDataType();
				FileLocation fileLoc = fileRep.getLocation();
				fileLoc.setFileSize(ingestUtilities.getFileSize(dataResource));

				if (ingestJob.getHost().booleanValue()) {
					// Copy to Piazza S3 bucket if hosted = true; If already in
					// S3, make
					// sure it's different than the Piazza S3; Depending on the
					// Type of file
					switch (fileLoc.getType()) {
					case S3FileStore.type:
						S3FileStore s3FS = (S3FileStore) fileLoc;
						if (!s3FS.getBucketName().equals(AMAZONS3_BUCKET_NAME)) {
							ingestUtilities.copyS3Source(dataResource);
							fileRep.setLocation(new S3FileStore(AMAZONS3_BUCKET_NAME, dataResource.getDataId() + "-"
									+ s3FS.getFileName(), s3FS.getFileSize(), s3FS.getDomainName()));
						}
						break;
					case FolderShare.type:
						ingestUtilities.copyS3Source(dataResource);
						break;
					}
				}
			}

			dataResource.metadata.createdBy = job.submitterUserName;
			dataResource.metadata.createdDate = job.submitted.toString();
<commit_msg>Null check to prevent crash<commit_after>/**
 * Copyright 2016, RadiantBlue Technologies, Inc.
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 **/
package ingest.messaging;

import ingest.event.IngestEvent;
import ingest.inspect.Inspector;
import ingest.utility.IngestUtilities;

import java.io.IOException;
import java.util.concurrent.Future;

import messaging.job.JobMessageFactory;
import messaging.job.WorkerCallback;
import model.data.DataResource;
import model.data.FileRepresentation;
import model.data.location.FileLocation;
import model.data.location.FolderShare;
import model.data.location.S3FileStore;
import model.job.Job;
import model.job.JobProgress;
import model.job.result.type.DataResult;
import model.job.result.type.ErrorResult;
import model.job.type.IngestJob;
import model.job.type.SearchMetadataIngestJob;
import model.response.PiazzaResponse;
import model.status.StatusUpdate;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import util.PiazzaLogger;
import util.UUIDFactory;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mongodb.MongoException;

/**
 * Worker class that handles a specific Ingest Job. The threads are managed by
 * the IngestThreadManager class.
 * 
 * @author Patrick.Doody & Sonny.Saniev & Russell.Orf
 * 
 */
@Component
public class IngestWorker {
	@Value("${SPACE}")
	private String SPACE;
	@Value("${workflow.event.id}")
	private String EVENT_ID;
	@Value("${workflow.url}")
	private String WORKFLOW_URL;
	@Value("${search.url}")
	private String SEARCH_URL;
	@Value("${workflow.endpoint}")
	private String WORKFLOW_ENDPOINT;
	@Value("${search.ingest.endpoint}")
	private String SEARCH_ENDPOINT;
	@Value("${vcap.services.pz-blobstore.credentials.bucket}")
	private String AMAZONS3_BUCKET_NAME;

	@Autowired
	private PiazzaLogger logger;
	@Autowired
	private Inspector inspector;
	@Autowired
	private IngestUtilities ingestUtilities;
	@Autowired
	private UUIDFactory uuidFactory;

	/**
	 * Creates a new Worker Thread for the specified Kafka Message containing an
	 * Ingest Job.
	 * 
	 * @param consumerRecord
	 *            The Kafka Message containing the Job.
	 * @param producer
	 *            The Kafka producer, used to send update messages
	 * @param callback
	 *            The callback that will be invoked when this Job has finished
	 *            processing (error or success, regardless)
	 */
	@Async
	public Future<DataResource> run(ConsumerRecord<String, String> consumerRecord, Producer<String, String> producer,
			WorkerCallback callback) {
		DataResource dataResource = null;

		try {
			// Log
			logger.log(String.format("Processing Data Load for Topic %s with Key %s", consumerRecord.topic(),
					consumerRecord.key()), PiazzaLogger.INFO);

			// Parse the Job from the Kafka Message
			ObjectMapper mapper = new ObjectMapper();
			Job job = mapper.readValue(consumerRecord.value(), Job.class);
			IngestJob ingestJob = (IngestJob) job.getJobType();
			// Get the description of the Data to be ingested
			dataResource = ingestJob.getData();

			// Assign a Resource ID to the incoming DataResource.
			if (dataResource.getDataId() == null) {
				String dataId = uuidFactory.getUUID();
				dataResource.setDataId(dataId);
			}

			// Log what we're going to Ingest
			logger.log(String.format(
					"Inspected Load Job; begin Loading Data %s of Type %s. Hosted: %s with Job ID of %s",
					dataResource.getDataId(), dataResource.getDataType().getType(), ingestJob.getHost().toString(),
					job.getJobId()), PiazzaLogger.INFO);

			// Update Status on Handling
			JobProgress jobProgress = new JobProgress(0);
			StatusUpdate statusUpdate = new StatusUpdate(StatusUpdate.STATUS_RUNNING, jobProgress);
			producer.send(JobMessageFactory.getUpdateStatusMessage(consumerRecord.key(), statusUpdate, SPACE));

			if (ingestJob.getData().getDataType() instanceof FileRepresentation) {
				FileRepresentation fileRep = (FileRepresentation) ingestJob.getData().getDataType();
				FileLocation fileLoc = fileRep.getLocation();
				fileLoc.setFileSize(ingestUtilities.getFileSize(dataResource));

				if (ingestJob.getHost().booleanValue()) {
					// Copy to Piazza S3 bucket if hosted = true; If already in
					// S3, make
					// sure it's different than the Piazza S3; Depending on the
					// Type of file
					switch (fileLoc.getType()) {
					case S3FileStore.type:
						S3FileStore s3FS = (S3FileStore) fileLoc;
						if (!s3FS.getBucketName().equals(AMAZONS3_BUCKET_NAME)) {
							ingestUtilities.copyS3Source(dataResource);
							fileRep.setLocation(new S3FileStore(AMAZONS3_BUCKET_NAME, dataResource.getDataId() + "-"
									+ s3FS.getFileName(), s3FS.getFileSize(), s3FS.getDomainName()));
						}
						break;
					case FolderShare.type:
						ingestUtilities.copyS3Source(dataResource);
						break;
					}
				}
			}

			if (dataResource.metadata != null) {
				dataResource.metadata.createdBy = job.submitterUserName;
				dataResource.metadata.createdDate = job.submitted.toString();
			}

{'fraction_non_alphanumeric': 0.10070714459887832, 'fraction_numerical': 0.0043891733723482075, 'mean_word_length': 4.439434129089301, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '407995', 'n_tokens_mistral': 3946, 'n_tokens_neox': 3551, 'n_words': 935}