#fixed code
                             public void next() {
		if (_waitingToEmit.isEmpty()) {

			fill();
		}

		while (true) {
			MessageAndOffset msgAndOffset = _waitingToEmit.pollFirst();
			
			if (msgAndOffset != null) {
				
				Long key = msgAndOffset.offset();
				Message msg = msgAndOffset.message();


				try {
					_lastEnquedOffset = key;
					if (_lastEnquedOffset >= _lastComittedOffset) {

						if (msg.payload() != null) {
														
							_receiver.store(msg.payload());
								 
							 LOG.info("Store for topic " + _topic + " for partition " + _partition.partition + " is : "+  _lastEnquedOffset);

						}
					}
				} catch (Exception e) {
					LOG.info("Process Failed for offset " + key + " for  "
							+ _partition + " for topic " + _topic
							+ " with Exception" + e.getMessage());
					e.printStackTrace();
				}
			}else{
				
				break;
			}
		}

		long now = System.currentTimeMillis();
		if ((_lastEnquedOffset > _lastComittedOffset) && ((now - _lastCommitMs) > _kafkaconfig._stateUpdateIntervalMs)) {
			commit();
			LOG.info("After commit , Waiting To Emit queue size is  "
					+ _waitingToEmit.size());
			_lastCommitMs = System.currentTimeMillis();
		}
	}
#vulnerable code 
                             public void next() {
		if (_waitingToEmit.isEmpty()) {

			fill();
		}

		while (true) {
			MessageAndOffset msgAndOffset = _waitingToEmit.pollFirst();
			
			if (msgAndOffset != null) {
				
				Long key = msgAndOffset.offset();
				Message msg = msgAndOffset.message();


				try {
					_lastEnquedOffset = key;
					if (_lastEnquedOffset > _lastComittedOffset) {

						if (msg.payload() != null) {
							
							synchronized (_receiver) {
								 _receiver.store(new String(Utils.toByteArray(msg.payload())));
							}
							 LOG.info("Store for topic " + _topic + " for partition " + _partition.partition + " is : "+  _lastEnquedOffset);

						}
					}
				} catch (Exception e) {
					LOG.info("Process Failed for offset " + key + " for  "
							+ _partition + " for topic " + _topic
							+ " with Exception" + e.getMessage());
					e.printStackTrace();
				}
			}else{
				
				break;
			}
		}

		long now = System.currentTimeMillis();
		if ((_lastEnquedOffset > _lastComittedOffset) && ((now - _lastCommitMs) > _kafkaconfig._stateUpdateIntervalMs)) {
			commit();
			LOG.info("After commit , Waiting To Emit queue size is  "
					+ _waitingToEmit.size());
			_lastCommitMs = System.currentTimeMillis();
		}
	} 
                            #location 25 
                            #vulnerability type THREAD_SAFETY_VIOLATION