#fixed code
                                @Override
    public void store(String url, Status status, Metadata metadata,
            Date nextFetch, Tuple tuple) throws Exception {

        String sha256hex = org.apache.commons.codec.digest.DigestUtils
                .sha256Hex(url);

        // need to synchronize: otherwise it might get added to the cache
        // without having been sent to ES
        synchronized (waitAck) {
            // check that the same URL is not being sent to ES
            List<Tuple> alreadySent = waitAck.getIfPresent(sha256hex);
            if (alreadySent != null && status.equals(Status.DISCOVERED)) {
                // if this object is discovered - adding another version of it
                // won't make any difference
                LOG.debug(
                        "Already being sent to ES {} with status {} and ID {}",
                        url, status, sha256hex);
                // ack straight away!
                super.ack(tuple, url);
                return;
            }
        }

        XContentBuilder builder = jsonBuilder().startObject();
        builder.field("url", url);
        builder.field("status", status);

        // check that we don't overwrite an existing entry
        // When create is used, the index operation will fail if a document
        // by that id already exists in the index.
        boolean create = status.equals(Status.DISCOVERED);

        builder.startObject("metadata");
        Iterator<String> mdKeys = metadata.keySet().iterator();
        while (mdKeys.hasNext()) {
            String mdKey = mdKeys.next();
            String[] values = metadata.getValues(mdKey);
            // periods are not allowed in ES2 - replace with %2E
            mdKey = mdKey.replaceAll("\\.", "%2E");
            builder.array(mdKey, values);
        }

        String partitionKey = partitioner.getPartition(url, metadata);
        if (partitionKey == null) {
            partitionKey = "_DEFAULT_";
        }

        // store routing key in metadata?
        if (StringUtils.isNotBlank(fieldNameForRoutingKey)
                && routingFieldNameInMetadata) {
            builder.field(fieldNameForRoutingKey, partitionKey);
        }

        builder.endObject();

        // store routing key outside metadata?
        if (StringUtils.isNotBlank(fieldNameForRoutingKey)
                && !routingFieldNameInMetadata) {
            builder.field(fieldNameForRoutingKey, partitionKey);
        }

        builder.field("nextFetchDate", nextFetch);

        builder.endObject();

        IndexRequest request = new IndexRequest(getIndexName(metadata));
        request.source(builder).id(sha256hex).create(create);

        if (doRouting) {
            request.routing(partitionKey);
        }

        synchronized (waitAck) {
            List<Tuple> tt = waitAck.getIfPresent(sha256hex);
            if (tt == null) {
                tt = new LinkedList<>();
                waitAck.put(sha256hex, tt);
            }
            tt.add(tuple);
            LOG.debug("Added to waitAck {} with ID {} total {}", url,
                    sha256hex, tt.size());
        }

        LOG.debug("Sending to ES buffer {} with ID {}", url, sha256hex);

        connection.getProcessor().add(request);
    }
#vulnerable code 
                                @Override
    public void store(String url, Status status, Metadata metadata,
            Date nextFetch, Tuple tuple) throws Exception {

        String sha256hex = org.apache.commons.codec.digest.DigestUtils
                .sha256Hex(url);

        // need to synchronize: otherwise it might get added to the cache
        // without having been sent to ES
        synchronized (waitAck) {
            // check that the same URL is not being sent to ES
            List<Tuple> alreadySent = waitAck.getIfPresent(sha256hex);
            if (alreadySent != null && status.equals(Status.DISCOVERED)) {
                // if this object is discovered - adding another version of it
                // won't make any difference
                LOG.debug(
                        "Already being sent to ES {} with status {} and ID {}",
                        url, status, sha256hex);
                // ack straight away!
                super.ack(tuple, url);
                return;
            }
        }

        XContentBuilder builder = jsonBuilder().startObject();
        builder.field("url", url);
        builder.field("status", status);

        // check that we don't overwrite an existing entry
        // When create is used, the index operation will fail if a document
        // by that id already exists in the index.
        boolean create = status.equals(Status.DISCOVERED);

        builder.startObject("metadata");
        Iterator<String> mdKeys = metadata.keySet().iterator();
        while (mdKeys.hasNext()) {
            String mdKey = mdKeys.next();
            String[] values = metadata.getValues(mdKey);
            // periods are not allowed in ES2 - replace with %2E
            mdKey = mdKey.replaceAll("\\.", "%2E");
            builder.array(mdKey, values);
        }

        String partitionKey = partitioner.getPartition(url, metadata);
        if (partitionKey == null) {
            partitionKey = "_DEFAULT_";
        }

        // store routing key in metadata?
        if (StringUtils.isNotBlank(fieldNameForRoutingKey)
                && routingFieldNameInMetadata) {
            builder.field(fieldNameForRoutingKey, partitionKey);
        }

        builder.endObject();

        // store routing key outside metadata?
        if (StringUtils.isNotBlank(fieldNameForRoutingKey)
                && !routingFieldNameInMetadata) {
            builder.field(fieldNameForRoutingKey, partitionKey);
        }

        builder.field("nextFetchDate", nextFetch);

        builder.endObject();

        IndexRequest request = new IndexRequest(getIndexName(metadata))
                .type(docType);
        request.source(builder).id(sha256hex).create(create);

        if (doRouting) {
            request.routing(partitionKey);
        }

        synchronized (waitAck) {
            List<Tuple> tt = waitAck.getIfPresent(sha256hex);
            if (tt == null) {
                tt = new LinkedList<>();
                waitAck.put(sha256hex, tt);
            }
            tt.add(tuple);
            LOG.debug("Added to waitAck {} with ID {} total {}", url,
                    sha256hex, tt.size());
        }

        LOG.debug("Sending to ES buffer {} with ID {}", url, sha256hex);

        connection.getProcessor().add(request);
    } 
                            #location 68 
                            #vulnerability type THREAD_SAFETY_VIOLATION