Send all cached logs to server

@return whether the logs were sent successfully
@throws IOException if closed or uploading batch fails
160638
public boolean sendBatch() throws IOException
  {
    if (isClosed)
    {
      throw new IOException("Telemetry connector is closed");
    }
    if (!isTelemetryEnabled())
    {
      return false;
    }

    LinkedList<TelemetryData> tmpList;
    synchronized (locker)
    {
      tmpList = this.logBatch;
      this.logBatch = new LinkedList<>();
    }

    if (session.isClosed())
    {
      throw new UnexpectedException("Session is closed when sending log");
    }

    if (!tmpList.isEmpty())
    {
      //session shared with JDBC
      String sessionToken = this.session.getSessionToken();

      HttpPost post = new HttpPost(this.telemetryUrl);
      post.setEntity(new StringEntity(logsToString(tmpList)));
      post.setHeader("Content-type", "application/json");
      post.setHeader("Authorization", "Snowflake Token=\"" + sessionToken +
                                      "\"");

      String response = null;

      try
      {
        response = HttpUtil.executeRequest(post, 1000, 0, null);
      }
      catch (SnowflakeSQLException e)
      {
        disableTelemetry(); // when got error like 404 or bad request, disable telemetry in this telemetry instance
        logger.error(
            "Telemetry request failed, " +
            "response: {}, exception: {}", response, e.getMessage());
        return false;
      }
    }
    return true;
  }
440553
public void flush()
        throws IOException
    {
        try
        {
            if (!_commited)
            {
                _commited=true;
                if (_commitObserver!=null)
                    _commitObserver.outputNotify(this,OutputObserver.__COMMITING,null);
            }
            
            wrapBuffer();
            
            // Add headers
            if (_httpMessageWriter.size()>0)
            {
                prewrite(_httpMessageWriter.getBuf(),0,_httpMessageWriter.size());
                _httpMessageWriter.resetWriter();
            }
            
            if (size()>0)
                writeTo(_out);
        }
        catch (IOException e)
        {
            throw new EOFException(e);
        }
        finally
        {
            reset(_preReserve);
        }
    }
1110281
boolean flush() {
    if (!dirty)
      return true;

    acquireExclusiveLock();
    try {

      final long timer = OProfiler.getInstance().startChrono();

      // FORCE THE WRITE OF THE BUFFER
      for (int i = 0; i < FORCE_RETRY; ++i) {
        try {
          buffer.force();
          dirty = false;
          break;
        } catch (Exception e) {
          OLogManager.instance().debug(this,
              "Cannot write memory buffer to disk. Retrying (" + (i + 1) + "/" + FORCE_RETRY + ")...");
          OMemoryWatchDog.freeMemory(FORCE_DELAY);
        }
      }

      if (dirty)
        OLogManager.instance().debug(this, "Cannot commit memory buffer to disk after %d retries", FORCE_RETRY);
      else
        OProfiler.getInstance().updateCounter("system.file.mmap.pagesCommitted", 1);

      OProfiler.getInstance().stopChrono("system.file.mmap.commitPages", timer);

      return !dirty;

    } finally {
      releaseExclusiveLock();
    }
  }
708260
public void flush() {
        List<BatchEntry> temp;

        bufferLock.lock();
        try {
            // Reset the last flush timestamp, even if the batch is empty or flush fails
            lastFlush = System.currentTimeMillis();

            // No-op if batch is empty
            if (batch == batchSize) {
                logger.trace("[{}] Batch empty, not flushing", name);
                return;
            }

            // Declare the batch empty, regardless of flush success/failure
            batch = batchSize;

            // If something goes wrong we still have a copy to recover.
            temp = buffer;
            buffer = new LinkedList<>();

        } finally {
            bufferLock.unlock();
        }

        long start = System.currentTimeMillis();
        try {
            flushTransactionLock.lock();
            start = System.currentTimeMillis();

            processBatch(temp);

            logger.trace("[{}] Batch flushed. Took {} ms, {} rows.", name, (System.currentTimeMillis() - start), temp.size());
        } catch (final Exception e) {
            if (this.maxFlushRetries > 0) {
                logger.warn(dev, "[{}] Error occurred while flushing. Retrying.", name, e);
            }

            boolean success = false;
            int retryCount;

            for (retryCount = 0; retryCount < this.maxFlushRetries && !success; retryCount++) {
                try {
                    Thread.sleep(this.flushRetryDelay);

                    // If the connection was established, we might need a rollback.
                    if (de.checkConnection() && de.isTransactionActive()) {
                        de.rollback();
                    }

                    processBatch(temp);

                    success = true;
                } catch (final InterruptedException ex) {
                    logger.debug("Interrupted while trying to flush batch. Stopping retries.");
                    Thread.currentThread().interrupt();
                    break;
                } catch (final Exception ex) {
                    logger.warn(dev, "[{}] Error occurred while flushing (retry attempt {}).", name, retryCount + 1, ex);
                }
            }

            if (!success) {
                try {
                    if (de.isTransactionActive()) {
                        de.rollback();
                    }
                } catch (final Exception ee) {
                    ee.addSuppressed(e);
                    logger.trace("[{}] Batch failed to check the flush transaction state", name, ee);
                }

                onFlushFailure(temp.toArray(new BatchEntry[temp.size()]));
                logger.error(dev, "[{}] Error occurred while flushing. Aborting batch flush.", name, e);
            } else {
                logger.trace("[{}] Batch flushed. Took {} ms, {} retries, {} rows.", name,
                        (System.currentTimeMillis() - start), retryCount, temp.size());
            }
        } finally {
            try {
                if (de.isTransactionActive()) {
                    de.rollback();
                }
            } catch (final Exception e) {
                logger.trace("[{}] Batch failed to check the flush transaction state", name, e);
            } finally {
                flushTransactionLock.unlock();
            }
        }
    }
103292
protected void drainCache() {
    // Route the tuples to correct places
    Map<Integer, List<HeronTuples.HeronTupleSet>> cache = tupleCache.getCache();

    if (!isSendTuplesToInstance(new LinkedList<Integer>(cache.keySet()))) {
      // Check whether we could send tuples
      return;
    }

    for (Map.Entry<Integer, List<HeronTuples.HeronTupleSet>> entry : cache.entrySet()) {
      int taskId = entry.getKey();
      for (HeronTuples.HeronTupleSet message : entry.getValue()) {
        sendInBound(taskId, message);
      }
    }

    // Reset the tupleCache
    tupleCache.clear();
  }
1314602
private void sendPipelineCheckMaster() throws IOException {
    if (urlParser.getHaMode() == HaMode.AURORA) {
      writer.startPacket(0);
      writer.write(COM_QUERY);
      writer.write(IS_MASTER_QUERY);
      writer.flush();
    }
  }
220917
public synchronized void rollLog() throws IOException {
        if (logStream != null) {
            this.logStream.flush();
            this.logStream = null;
            oa = null;
        }
    }
1104868
private boolean performFlushIfRequired() throws IOException {
        if (anyAreSet(state, FLUSHING_BUFFER)) {
            final ByteBuffer[] bufs = new ByteBuffer[additionalBuffer == null ? 1 : 2];
            long totalLength = 0;
            bufs[0] = currentBuffer.getBuffer();
            totalLength += bufs[0].remaining();
            if (additionalBuffer != null) {
                bufs[1] = additionalBuffer;
                totalLength += bufs[1].remaining();
            }
            if (totalLength > 0) {
                long total = 0;
                long res = 0;
                do {
                    res = next.write(bufs, 0, bufs.length);
                    total += res;
                    if (res == 0) {
                        return false;
                    }
                } while (total < totalLength);
            }
            additionalBuffer = null;
            currentBuffer.getBuffer().clear();
            state = state & ~FLUSHING_BUFFER;
        }
        return true;
    }
756770
public function send() {
    if (!isset($this->files[$this->sent])) {
      return FALSE;
    }

    $file = $this->files[$this->sent];

    // Local header
    $local = $file->genLocal(self::ZIP_VERSION, $this->dosStamp);
    $this->output($local);

    // Body
    $file->hasDeflation() ? $this->outputDeflated($file) : $this->outputRaw($file);

    // Data descriptor
    $dataDescriptor = $file->isSmooth() ? $file->genDataDescriptor() : '';
    $this->output($dataDescriptor);

    $file->setLocalOffset($this->localOffset);
    $this->localOffset += strlen($local) + $file->getDeflatedSize() + strlen($dataDescriptor);

    // Central header (not output, see flush)
    $file->genCentral(self::ZIP_VERSION, $this->dosStamp);
    $this->centralLength += strlen($file->getCentral());

    $this->sent++;
    return TRUE;
  }
419758
public void flush() throws IOException {
        if(closed) {
            throw new IllegalStateException("Can't flush packets on a closed stream!");
        }

        // Write in one go
        OggPage[] pages = buffer.toArray(new OggPage[buffer.size()]); 
        file.writePages(pages);

        // Get ready for next time!
        buffer.clear();
    }
642013
protected void submitTraces() {
        if (!traces.isEmpty()) {
            // Locally store list and create new list for subsequent traces
            List<Trace> toSend = traces;
            traces = new ArrayList<>(batchSize + 1);

            executor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        tracePublisher.publish(tenantId, toSend);
                    } catch (Exception e) {
                        // TODO: Retain for retry
                        log.log(Level.SEVERE, "Failed to publish traces", e);
                    }
                }
            });
        }
    }
222974
private void flushBytes() throws IOException
	{
		if (com.ibm.ejs.ras.TraceComponent.isAnyTracingEnabled()&&logger.isLoggable (Level.FINE)) {  //306998.15
			logger.logp(Level.FINE, CLASS_NAME,"flushBytes", "flushBytes");
        }
		if (!committed)
		{
			if (!_hasFlushed && obs != null)
			{
				_hasFlushed = true;
				if (obsList!=null){
					for (int i=0;i<obsList.size();i++){
						obsList.get(i).alertFirstFlush();
					}
				}
				else {
				obs.alertFirstFlush();
			}
		}
		}
		committed = true;
		if (count > 0)
		{
			writeOut(buf, 0, count);
// PK34562			out.flush();
			count = 0;
		}
		// PK34562 Start
//  		else if ( count == 0 && out != null) {
//			out.flush();
//		}
		else if (count == 0)
		{
	        if(response != null && response.getFlushMode())
	        {
	        	if (com.ibm.ejs.ras.TraceComponent.isAnyTracingEnabled()&&logger.isLoggable (Level.FINE)) {  //PK34562
	    			logger.logp(Level.FINE, CLASS_NAME,"flushBytes", "Count 0 still flush mode is true , forceful flush");
	            }
	            response.flushBufferedContent();
	        } else if (com.ibm.ejs.ras.TraceComponent.isAnyTracingEnabled()&&logger.isLoggable (Level.FINE)) {  //PK34562
				logger.logp(Level.FINE, CLASS_NAME,"flushBytes", "flush mode is false");
	        }	 
		}
		// PK34562 End
	}
411065
public synchronized boolean log(WsByteBuffer data) {
        if (null == data) {
            // return failure
            return false;
        }
        // if we've stopped then there is no worker to hand this to; however, the
        // caller does not expect to have to release buffers handed to the
        // logger so do that here first
        if (State.RUNNING != this.state) {
            data.release();
            return false;
        }
        int length = data.remaining();
        if (isOverFileLimit(length)) {
            rotate();
        }
        boolean rc = true;
        int bytesWritten = 0;
        try {
            ByteBuffer buffer = data.getWrappedByteBuffer();
            while (bytesWritten < length) {
                bytesWritten += this.myChannel.write(buffer);
            }
        } catch (Exception e) {
            FFDCFilter.processException(e, getClass().getName() + ".log", "166", this);
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
                Tr.event(tc, getFileName() + ": error writing to log; " + e);
            }
            rc = false;
        } finally {
            data.release();
        }
        return rc;
    }
59644
public void flushTxLog() {
        synchronized (flushCompleteSignal) {
            BufferName enqueueBuffer;
            synchronized (txLogToVEBuf.getPot()) {
                // Enqueue the backup task
                enqueueBuffer = txLogToVEBuf.getEnqueueBuffer(); 
                FlushNowJob flushJob = new FlushNowJob();
                txLogToVEBuf.addAndExecuteWork(flushJob);
            }
            
            try {
                do {
                    log.debug("Waiting for flush to complete in buffer {}", enqueueBuffer);
                    flushCompleteSignal.wait();
                    log.debug("Flush complete in buffer {}", lastFlushedBuffer);
                } while (lastFlushedBuffer != enqueueBuffer);
            } catch (InterruptedException e) {
                BitsyException toThrow = new BitsyException(BitsyErrorCodes.FLUSH_INTERRUPTED, "Exception while waiting for a transaction-log flush to be performed", e);

                log.error("Error while flushing the transaction log", toThrow);

                throw toThrow;
            }
        }        
    }
303870
public void dump() throws InvocationException {
		if (_updated) {
			synchronized (_lock) {
				if (!_updated)
					return;

				List<LogMessage> messages = _cache;
				_cache = new ArrayList<LogMessage>();

				save(messages);

				_updated = false;
				_lastDumpTime = System.currentTimeMillis();
			}
		}
	}
697696
public synchronized void writeLog(String log) throws IOException {
		if (mFile.length() < mFileMaxSize) {
			writeLogToFile(log, mFile, true);
		} else {
			rotateLogFiles();
			createNewFile();
			writeLogToFile(log, mFile, false);
		}
	}
941159
@Override
  public void flush() throws IOException
  {
    if (_os == null || ! _needsFlush)
      return;

    _needsFlush = false;
    try {
      _os.flush();
    } catch (IOException e) {
      try {
        close();
      } catch (IOException e1) {
      }
      
      throw ClientDisconnectException.create(e);
    }
  }