Skip to content

Commit

Permalink
events for unassigned users are not processed but return 404 'assignm…
Browse files Browse the repository at this point in the history 
…ent not found'
  • Loading branch information
@shoeffner
shoeffner committed Oct 12, 2016
1 parent 808141f commit dc71670
Show file tree
Hide file tree
Showing 4 changed files with 36 additions and 26 deletions.
55 changes: 33 additions & 22 deletions modules/event/src/main/java/com/intuit/wasabi/events/impl/EventsImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,21 +19,23 @@
import com.google.inject.name.Named;
import com.intuit.wasabi.analyticsobjects.Event;
import com.intuit.wasabi.analyticsobjects.EventList;
import com.intuit.wasabi.events.EventIngestionExecutor;
import com.intuit.wasabi.assignment.Assignments;
import com.intuit.wasabi.assignmentobjects.Assignment;
import com.intuit.wasabi.eventobjects.EventEnvelopePayload;
import com.intuit.wasabi.assignmentobjects.User;
import com.intuit.wasabi.database.TransactionFactory;
import com.intuit.wasabi.eventobjects.EventEnvelopePayload;
import com.intuit.wasabi.events.EventIngestionExecutor;
import com.intuit.wasabi.events.Events;
import com.intuit.wasabi.events.EventsMBean;
import com.intuit.wasabi.exceptions.AssignmentNotFoundException;
import com.intuit.wasabi.experimentobjects.Application;
import com.intuit.wasabi.experimentobjects.Context;
import com.intuit.wasabi.experimentobjects.Experiment;
import org.slf4j.Logger;

import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
Expand All @@ -50,22 +52,22 @@
*/
public class EventsImpl implements Events, EventsMBean {

private static final Logger LOGGER = getLogger(EventsImpl.class);
protected static final String MYSQL = "mysql";
private Assignments assignments;
private TransactionFactory transactionFactory;
private LinkedBlockingQueue mysqlQueue = new LinkedBlockingQueue<>();
private ThreadPoolExecutor mysqlExecutor;
private static final Logger LOGGER = getLogger(EventsImpl.class);
/**
* Executors to ingest event data to real time ingestion system.
*/
protected Map<String, EventIngestionExecutor> eventIngestionExecutors;
private Assignments assignments;
private TransactionFactory transactionFactory;
private LinkedBlockingQueue mysqlQueue = new LinkedBlockingQueue<>();
private ThreadPoolExecutor mysqlExecutor;

@Inject
public EventsImpl(Map<String, EventIngestionExecutor> eventIngestionExecutors,
final @Named("executor.threadpool.size") Integer threadPoolSize,
final Assignments assignments,
final TransactionFactory transactionFactory) {
final @Named("executor.threadpool.size") Integer threadPoolSize,
final Assignments assignments,
final TransactionFactory transactionFactory) {
super();
this.eventIngestionExecutors = eventIngestionExecutors;
this.transactionFactory = transactionFactory;
Expand All @@ -82,25 +84,33 @@ public void recordEvents(Application.Name applicationName,
Experiment.Label experimentLabel, User.ID userID, EventList events, Set<Context> contextSet) {
Map<Context, Assignment> assignmentHashMap = getAssignments(userID, applicationName, experimentLabel, contextSet);

if (events.getEvents()
.parallelStream()
.filter(event -> Objects.nonNull(assignmentHashMap.get(event.getContext())))
.count()
!= events.getEvents().size()) {
throw new AssignmentNotFoundException(userID, applicationName, experimentLabel);
}

for (Event event : events.getEvents()) {
Assignment assignment = assignmentHashMap.get(event.getContext());
if (assignment != null) {
postEventToMysql(assignment, event);
ingestEventToRealTimeSystems(applicationName, experimentLabel, event, assignment);
}
postEventToMysql(assignment, event);
ingestEventToRealTimeSystems(applicationName, experimentLabel, event, assignment);
}
}

// This method ingests event to real time ingestion systems.
/**
* This method ingests event to real time ingestion systems.
*/
private void ingestEventToRealTimeSystems(Application.Name applicationName, Experiment.Label experimentLabel, Event event,
Assignment assignment) {
Assignment assignment) {
for (String name : eventIngestionExecutors.keySet()) {
eventIngestionExecutors.get(name).execute(new EventEnvelopePayload(applicationName, experimentLabel, assignment, event));
}
}

protected Map<Context, Assignment> getAssignments(User.ID userID, Application.Name applicationName,
Experiment.Label experimentLabel, Set<Context> contextSet) {
Experiment.Label experimentLabel, Set<Context> contextSet) {
Map<Context, Assignment> assignmentHashMap = new HashMap<>();
for (Context context : contextSet) {
Assignment assignment = assignments.getAssignment(
Expand All @@ -121,13 +131,14 @@ private void postEventToMysql(Assignment assignment, Event event) {

/**
* Helper method to instantiate events envelope
*
* @param assignment
* @param event
* @return
*/
protected EventsEnvelope makeEventEnvelope(Assignment assignment, Event event) {
return new EventsEnvelope(assignment, event, transactionFactory.newTransaction());
}
protected EventsEnvelope makeEventEnvelope(Assignment assignment, Event event) {
return new EventsEnvelope(assignment, event, transactionFactory.newTransaction());
}

/**
* {@inheritDoc}
Expand All @@ -143,7 +154,7 @@ public Map<String, Integer> queuesLength() {
queueLengthMap.put(MYSQL, new Integer(mysqlQueue.size()));
for (String name : eventIngestionExecutors.keySet()) {
queueLengthMap.put(name.toLowerCase(), new Integer(eventIngestionExecutors.get(name).queueLength()));
}
}
return queueLengthMap;
}

Expand All @@ -154,5 +165,5 @@ public Map<String, Integer> queuesLength() {
public int getQueueSize() {
// FIXME: is this MBean method really used??
return mysqlQueue.size();
}
}
}
3 changes: 1 addition & 2 deletions ...unctional-test/src/main/java/com/intuit/wasabi/tests/model/factory/ExperimentFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,8 +52,7 @@ public static Experiment createExperiment() {
TestUtils.currentTimeString(),
TestUtils.relativeTimeString(42), 1)
.setCreatorID(USER_ID)
.setDescription("A sample Experiment description.")
.setState(Constants.EXPERIMENT_STATE_DRAFT);
.setDescription("A sample Experiment description.");
}

/**
Expand Down
2 changes: 1 addition & 1 deletion ...l-test/src/main/java/com/intuit/wasabi/tests/service/PastDatedExperimentCreationTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ public void createExperimentPastStartDateFutureEndDate() {
experiment.startTime = TestUtils.relativeTimeString(-10);
experiment.endTime = TestUtils.relativeTimeString(5);
Experiment exp = postExperiment(experiment);
ModelAssert.assertEqualModelItems(exp, experiment, new DefaultNameExclusionStrategy("id", "creationTime", "modificationTime", "ruleJson"));
ModelAssert.assertEqualModelItems(exp, experiment, new DefaultNameExclusionStrategy("id", "state", "creationTime", "modificationTime", "ruleJson"));
}

}
2 changes: 1 addition & 1 deletion .../java/com/intuit/wasabi/tests/service/assignment/EventSubmissionBeforeAssignmentTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ public void createImpression() {

//create an event of type IMPRESSION and post it to the event endpoint
Event impression = EventFactory.createImpression();
postEvent(impression, experiment, user, HttpStatus.SC_BAD_REQUEST);
postEvent(impression, experiment, user, HttpStatus.SC_NOT_FOUND);
}


Expand Down

0 comments on commit dc71670

Please sign in to comment.