#vulnerable code 
                                @Override
    public void send(Event event, int streamIndex) {
        try {
            long sequenceNo = ringBuffer.next();
            try {
                IndexedEventFactory.IndexedEvent existingEvent = ringBuffer.get(sequenceNo);
                existingEvent.setEvent(event);
                existingEvent.setStreamIndex(streamIndex);
            } finally {
                eventSizeInDisruptor.incrementAndGet();
                ringBuffer.publish(sequenceNo);    //Todo fix this for array of events
            }
        } catch (NullPointerException e) {
            throw new ExecutionPlanRuntimeException("Execution Plan:" + executionPlanContext.getName() + " not " +
                    "initialised yet! Run executionPlanRuntime.start();", e);
        }

    } 
                            #location 11 
                            #vulnerability type THREAD_SAFETY_VIOLATION
#fixed code
                                @Override
    public void send(Event event, int streamIndex) {
        inputProcessor.send(event, streamIndex);
//        try {
//            long sequenceNo = ringBuffer.next();
//            try {
//                IndexedEventFactory.IndexedEvent existingEvent = ringBuffer.get(sequenceNo);
//                existingEvent.setEvent(event);
//                existingEvent.setStreamIndex(streamIndex);
//            } finally {
//                eventSizeInDisruptor.incrementAndGet();
//                ringBuffer.publish(sequenceNo);    //Todo fix this for array of events
//            }
//        } catch (NullPointerException e) {
//            throw new ExecutionPlanRuntimeException("Execution Plan:" + executionPlanContext.getName() + " not " +
//                    "initialised yet! Run executionPlanRuntime.start();", e);
//        }

    }