@Override public void run(){
  List<Span> dequeuedSpans=new ArrayList<Span>(maxSpanBatchSize);
  long errorCount=0;
  while (running.get() || queue.size() > 0) {
    Span firstSpan=null;
    try {
      firstSpan=queue.poll(1,TimeUnit.SECONDS);
      if (firstSpan != null) {
        dequeuedSpans.add(firstSpan);
        queue.drainTo(dequeuedSpans,maxSpanBatchSize - 1);
      }
    }
 catch (    InterruptedException ie) {
    }
    startClient();
    if (dequeuedSpans.isEmpty()) {
      continue;
    }
    try {
      List<Event> events=new ArrayList<Event>(dequeuedSpans.size());
      for (      Span span : dequeuedSpans) {
        Map<String,String> headers=new HashMap<String,String>();
        headers.put("SpanId",span.toString());
        headers.put("TracerId",span.getTracerId());
        headers.put("Description",span.getDescription());
        String body=span.toJson();
        Event evt=EventBuilder.withBody(body,Charset.forName("UTF-8"),headers);
        events.add(evt);
      }
      flumeClient.appendBatch(events);
      dequeuedSpans.clear();
      errorCount=0;
    }
 catch (    Exception e) {
      errorCount+=1;
      if (errorCount < MAX_ERRORS) {
        try {
          queue.addAll(dequeuedSpans);
        }
 catch (        IllegalStateException ex) {
          LOG.error("Drop " + dequeuedSpans.size() + " span(s) because writing to HBase failed.");
        }
      }
      closeClient();
      try {
        Thread.sleep(500);
      }
 catch (      InterruptedException e1) {
      }
    }
  }
  closeClient();
}

@ Override public void run ( ) { List < Span > dequeuedSpans = new ArrayList < Span > ( maxSpanBatchSize ) ; long errorCount = 0 ; while ( running . get ( ) || queue . size ( ) > 0 ) { Span firstSpan = null ; try { firstSpan = queue . poll ( 1 , TimeUnit . SECONDS ) ; if ( firstSpan != null ) { dequeuedSpans . add ( firstSpan ) ; queue . drainTo ( dequeuedSpans , maxSpanBatchSize - 1 ) ; } } catch ( InterruptedException ie ) { } startClient ( ) ; if ( dequeuedSpans . isEmpty ( ) ) { continue ; } try { List < Event > events = new ArrayList < Event > ( dequeuedSpans . size ( ) ) ; for ( Span span : dequeuedSpans ) { Map < String , String > headers = new HashMap < String , String > ( ) ; headers . put ( "SpanId" , span . toString ( ) ) ; headers . put ( "TracerId" , span . getTracerId ( ) ) ; headers . put ( "Description" , span . getDescription ( ) ) ; String body = span . toJson ( ) ; Event evt = EventBuilder . withBody ( body , Charset . forName ( "UTF-8" ) , headers ) ; events . add ( evt ) ; } flumeClient . appendBatch ( events ) ; dequeuedSpans . clear ( ) ; errorCount = 0 ; } catch ( Exception e ) { errorCount += 1 ; if ( errorCount < MAX_ERRORS ) { try { queue . addAll ( dequeuedSpans ) ; } catch ( IllegalStateException ex ) { LOG . error ( "Drop " + dequeuedSpans . size ( ) + " span(s) because writing to HBase failed." ) ; } } closeClient ( ) ; try { Thread . sleep ( 500 ) ; } catch ( InterruptedException e1 ) { } } } closeClient ( ) ; }
this runnable sends a htrace span to the flume .
