#vulnerable code 
                                public static OutputCallback constructOutputCallback(OutputStream outStream, StreamDefinition outputStreamDefinition,
                                                         ExecutionPlanContext executionPlanContext) {
        String id = outStream.getId();

        //Construct CallBack
        if (outStream instanceof InsertIntoStream) {
            EventTable eventTable = executionPlanContext.getEventTableMap().get(id);
            if (eventTable != null) {
                DefinitionParserHelper.validateOutputStream(outputStreamDefinition, eventTable.getTableDefinition());
                return new InsertIntoTableCallback(eventTable, outputStreamDefinition);
            } else {
                return new InsertIntoStreamCallback(outputStreamDefinition);
            }
        } else if (outStream instanceof DeleteStream) {
            EventTable eventTable = executionPlanContext.getEventTableMap().get(id);
            if (eventTable != null) {
                DefinitionParserHelper.validateOutputStream(outputStreamDefinition, eventTable.getTableDefinition());
                MetaStreamEvent matchingMetaStreamEvent = new MetaStreamEvent();
                matchingMetaStreamEvent.setTableEvent(true);
                TableDefinition tableDefinition = TableDefinition.id("");
                for (Attribute attribute : outputStreamDefinition.getAttributeList()) {
                    matchingMetaStreamEvent.addOutputData(attribute);
                    tableDefinition.attribute(attribute.getName(), attribute.getType());
                }
                matchingMetaStreamEvent.setInputDefinition(tableDefinition);
                Finder finder = eventTable.constructFinder(((DeleteStream) outStream).getOnDeleteExpression(), matchingMetaStreamEvent, executionPlanContext, null, 0);
                return new DeleteTableCallback(eventTable, finder);
            } else {
                throw new DefinitionNotExistException("Event table with id :" + id + " does not exist");
            }
        } else if (outStream instanceof UpdateStream) {
            EventTable eventTable = executionPlanContext.getEventTableMap().get(id);
            if (eventTable != null) {
                DefinitionParserHelper.validateOutputStream(outputStreamDefinition, eventTable.getTableDefinition());
                MetaStateEvent metaStateEvent = createMetaStateEvent(outputStreamDefinition, eventTable);
                Finder finder = eventTable.constructFinder(((UpdateStream) outStream).getOnUpdateExpression(), metaStateEvent, executionPlanContext, null, 0);
                return new UpdateTableCallback(eventTable, finder, outputStreamDefinition);
            } else {
                throw new DefinitionNotExistException("Event table with id :" + id + " does not exist");
            }
        } else {
            throw new ExecutionPlanCreationException(outStream.getClass().getName() + " not supported");
        }

    } 
                            #location 21 
                            #vulnerability type NULL_DEREFERENCE
#fixed code
                                public static OutputCallback constructOutputCallback(OutputStream outStream, StreamDefinition outputStreamDefinition,
                                                         ExecutionPlanContext executionPlanContext) {
        String id = outStream.getId();

        //Construct CallBack
        if (outStream instanceof InsertIntoStream) {
            EventTable eventTable = executionPlanContext.getEventTableMap().get(id);
            if (eventTable != null) {
                DefinitionParserHelper.validateOutputStream(outputStreamDefinition, eventTable.getTableDefinition());
                return new InsertIntoTableCallback(eventTable, outputStreamDefinition);
            } else {
                return new InsertIntoStreamCallback(outputStreamDefinition);
            }
        } else if (outStream instanceof DeleteStream || outStream instanceof UpdateStream) {
            EventTable eventTable = executionPlanContext.getEventTableMap().get(id);
            if (eventTable != null) {

                TableDefinition eventTableDefinition = eventTable.getTableDefinition();
                for (Attribute attribute : outputStreamDefinition.getAttributeList()) {
                    if (!eventTableDefinition.getAttributeList().contains(attribute)) {
                        throw new ExecutionPlanCreationException("Attribute " + attribute + " does not exist on Event Table " + eventTableDefinition);
                    }
                }

                MetaStreamEvent matchingMetaStreamEvent = new MetaStreamEvent();
                matchingMetaStreamEvent.setTableEvent(true);
                TableDefinition matchingTableDefinition = TableDefinition.id("");
                for (Attribute attribute : outputStreamDefinition.getAttributeList()) {
                    matchingMetaStreamEvent.addOutputData(attribute);
                    matchingTableDefinition.attribute(attribute.getName(), attribute.getType());
                }
                matchingMetaStreamEvent.setInputDefinition(matchingTableDefinition);
                if (outStream instanceof DeleteStream) {
                    Finder finder = eventTable.constructFinder(((DeleteStream) outStream).getOnDeleteExpression(), matchingMetaStreamEvent, executionPlanContext, null, 0);
                    return new DeleteTableCallback(eventTable, finder);
                } else {
                    Finder finder = eventTable.constructFinder(((UpdateStream) outStream).getOnUpdateExpression(), matchingMetaStreamEvent, executionPlanContext, null, 0);
                    return new UpdateTableCallback(eventTable, finder, matchingTableDefinition);
                }
            } else {
                throw new DefinitionNotExistException("Event table with id :" + id + " does not exist");
            }
        } else {
            throw new ExecutionPlanCreationException(outStream.getClass().getName() + " not supported");
        }

    }