__tablename__ = "task_instance"

    task_id = Column(String(ID_LEN), primary_key=True)
    dag_id = Column(String(ID_LEN), primary_key=True)
    execution_date = Column(DateTime, primary_key=True)
    start_date = Column(DateTime)
    end_date = Column(DateTime)
    duration = Column(Integer)
    state = Column(String(20))
    try_number = Column(Integer)
    hostname = Column(String(1000))
    unixname = Column(String(1000)) 

class BaseSensorOperator(BaseOperator):

   @apply_defaults
   def __init__(
           self,
           poke_interval=60,
           timeout=60*60*24*7,
           *args, **kwargs):
       super(BaseSensorOperator, self).__init__(*args, **kwargs)
       self.poke_interval = poke_interval
       self.timeout = timeout

       # Since a sensor pokes in a loop, no need for higher level retries
       self.retries = 0

   def poke(self):
       '''
       Function that the sensors defined while deriving this class should
       override.
       '''
       raise Exception('Override me.')

   def execute(self, execution_date):
       started_at = datetime.now()
       while not self.poke():
           sleep(self.poke_interval)
           if (datetime.now() - started_at).seconds > self.timeout:
               raise Exception('Snap. Time is OUT.')
       logging.info("Success criteria met. Exiting.")

while True:
    for dag in dags:
        for task in dag.tasks:
            if task.is_runnable():
                execute()....

    def is_runnable(self):
        """
        Returns a boolean on whether the task instance has met all dependencies
        and is ready to run. It considers the task's state, the state
        of its dependencies, depends_on_past and makes sure the execution
        isn't in the future.
        """
        if self.execution_date > datetime.now() - self.task.schedule_interval:
            return False
        elif self.state == State.UP_FOR_RETRY and not self.ready_for_retry():
            return False
        elif self.state in State.runnable() and self.are_dependencies_met():
            return True
        else:
            return False

ERROR [SVC=A][trace=a1b2c3] Failed to process order cause:xxxx
ERROR [SVC=F][trace=a1b2c3] Failed to process order cause:xxxx
ERROR [SVC=G][trace=a1b2c3] Failed to process order cause:xxxx
ERROR [SVC=B][trace=a1b2c3] Failed to process order cause:DB timeout

def start(): StreamingQuery = {
     ...
    if (source == "memory") {
    ...
    } else if (source == "foreach") {
    ...  
    } else {
      val ds = DataSource.lookupDataSource(source, df.sparkSession.sessionState.conf)
      val disabledSources = df.sparkSession.sqlContext.conf.disabledV2StreamingWriters.split(",")
      val sink = ds.newInstance() match {
        case w: StreamWriteSupport if !disabledSources.contains(w.getClass.getCanonicalName) => w
        case _ =>
          val ds = DataSource(
            df.sparkSession,
            className = source,
            options = extraOptions.toMap,
            partitionColumns = normalizedParCols.getOrElse(Nil))
          ds.createSink(outputMode)
      }
...

       case sources =>
          // There are multiple registered aliases for the input. If there is single datasource
          // that has "org.apache.spark" package in the prefix, we use it considering it is an
          // internal datasource within Spark.
          val sourceNames = sources.map(_.getClass.getName)
          val internalSources = sources.filter(_.getClass.getName.startsWith("org.apache.spark"))
          if (internalSources.size == 1) {
            logWarning(s"Multiple sources found for $provider1 (${sourceNames.mkString(", ")}), " +
              s"defaulting to the internal datasource (${internalSources.head.getClass.getName}).")
            internalSources.head.getClass
          } else {
            throw new AnalysisException(s"Multiple sources found for $provider1 " +
              s"(${sourceNames.mkString(", ")}), please specify the fully qualified class name.")
          }

class DebugWriter(schema: StructType, options: DataSourceOptions) extends StreamWriter {

  assert(SparkSession.getActiveSession.isDefined)
  protected val spark = SparkSession.getActiveSession.get

  implicit val formats = org.json4s.DefaultFormats

  override def commit(epochId: Long, messages: Array[WriterCommitMessage]): Unit = {

    printRows(messages, schema, s"Batch: $epochId")
  }

  override def abort(epochId: Long, messages: Array[WriterCommitMessage]): Unit = {}

  override def createWriterFactory(): DataWriterFactory[Row] = PackedRowWriterFactory

  protected def printRows(
                           commitMessages: Array[WriterCommitMessage],
                           schema: StructType,
                           printMessage: String): Unit = {
    val rows = commitMessages.collect {
      case PackedRowCommitMessage(rs) => rs
    }.flatten


    val sample = spark
      .createDataFrame(rows.take(10).toList.asJava, schema)
      .toJSON
      .collect()

    val message = Serialization.writePretty(Map("message" -> sample))

    // scalastyle:off println
    println("-------------------------------------------")
    println(message)
    println("-------------------------------------------")
    // scalastyle:off println
  }
}

for (int i: list) {
    if (queue.size()  < k) {
        queue.offer(i);
    } else {
        if (i > queue.peek()) {
            queue.poll();
            queue.offer(i);
       }
    }
}

import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class TtlMap<K, V> {
    private final int N; // count of bucket
    private final int ttl;
    private volatile int curBucket = -1;
    private volatile int prevBucket = -1;
    private final List<Map<K, V>> maps;
    private final ScheduledExecutorService cleanService;
    public TtlMap(int ttl) {
        this.N = 60 / ttl;
        this.ttl = ttl;
        this.maps = new ArrayList<>(this.N);
        for (int i = 0; i < N; i++) {
            maps.add(i, null);
        }
        cleanService = Executors.newSingleThreadScheduledExecutor();
        cleanService.scheduleAtFixedRate(
                () -> cleanOldData()
                , 0, ttl, TimeUnit.SECONDS);
    }

    public void put(K key, V value) {
        long seconds = System.currentTimeMillis()/1000;
        int bucket = (int) seconds % N;
        if (bucket != curBucket) {
            // move forward the curBucket
            curBucket = bucket;
            prevBucket = getPrev(curBucket);
        }
        if (maps.get(curBucket) == null) {
            maps.add(curBucket, new HashMap<>());
        }
        maps.get(curBucket).put(key, value);
    }

    public V get(K key) {
        if (curBucket == -1 || prevBucket == -1) return null;
        if (maps.get(curBucket) != null && maps.get(curBucket).containsKey(key)) {
            return maps.get(curBucket).get(key);
        }
        if (maps.get(prevBucket) != null && maps.get(prevBucket).containsKey(key)) {
            return maps.get(prevBucket).get(key);
        }
        return null;
    }

    private int getPrev(int i) {
        return i != 0? i - 1: N - 1;
    }

    private void cleanOldData() {
        for (int i = 0; i < N; i++) {
            // bypass the current and prev bucket
            if (i == curBucket || i == prevBucket) continue;
            this.maps.set(i, null);
        }
    }

    public void stop() {
        cleanService.shutdown();
        for (int i = 0; i < N; i++) {
            // help GC
            maps.set(i, null);
        }
    }

    public static void main(String[] args) throws Exception {
        TtlMap<String, String> map = new TtlMap<>(5);
        map.put("key1", "value1");
        map.put("key2", "value2");
        System.out.println("0 seconds, get the key1 and key2");
        System.out.println("key1 --> " + map.get("key1") + "\tkey2--> " + map.get("key2"));
        Thread.sleep(5000);
        System.out.println("5 seconds, update key2");
        map.put("key2", "value2");
        Thread.sleep(2000);
        System.out.println("7 seconds, get the key1 and key2");
        System.out.println("key1 --> " + map.get("key1") + "\tkey2--> " + map.get("key2"));
        map.put("key3", "value3");
        Thread.sleep(5000);
        System.out.println("12 seconds");
        System.out.println("key1 --> " + map.get("key1") + "\tkey2--> " + map.get("key2"));
        map.stop();
    }
}

0 seconds, get the key1 and key2
key1 --> value1	key2--> value2
5 seconds, update key2
7 seconds, get the key1 and key2
key1 --> null	key2--> value2
12 seconds
key1 --> null	key2--> null

        class Flush implements Runnable {
    
            @Override
            public void run() {
                synchronized (BulkProcessor.this) {
                    if (closed) {
                        return;
                    }
                    if (bulkRequest.numberOfActions() == 0) {
                        return;
                    }
                    execute();
                }
            }
        }

        public void execute(BulkRequest bulkRequest, long executionId) {
            Runnable toRelease = () -> {};
            boolean bulkRequestSetupSuccessful = false;
            try {
                listener.beforeBulk(executionId, bulkRequest);
                semaphore.acquire();
                toRelease = semaphore::release;
                CountDownLatch latch = new CountDownLatch(1);
                retry.withBackoff(consumer, bulkRequest, new ActionListener<BulkResponse>() {
                    @Override
                    public void onResponse(BulkResponse response) {
                        try {
                            listener.afterBulk(executionId, bulkRequest, response);
                        } finally {
                            semaphore.release();
                            latch.countDown();
                        }
                    }
    
                    @Override
                    public void onFailure(Exception e) {
                        try {
                            listener.afterBulk(executionId, bulkRequest, e);
                        } finally {
                            semaphore.release();
                            latch.countDown();
                        }
                    }
                });
                bulkRequestSetupSuccessful = true;
                if (concurrentRequests == 0) {
                    latch.await();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                logger.info(() -> new ParameterizedMessage("Bulk request {} has been cancelled.", executionId), e);
                listener.afterBulk(executionId, bulkRequest, e);
            } catch (Exception e) {
                logger.warn(() -> new ParameterizedMessage("Failed to execute bulk request {}.", executionId), e);
                listener.afterBulk(executionId, bulkRequest, e);
            } finally {
                if (bulkRequestSetupSuccessful == false) {  // if we fail on client.bulk() release the semaphore
                    toRelease.run();
                }
            }
        }

            public void execute(BulkRequest bulkRequest) {
                this.currentBulkRequest = bulkRequest;
                consumer.accept(bulkRequest, this);
            }
			
		private void retry(BulkRequest bulkRequestForRetry) {
            assert backoff.hasNext();
            TimeValue next = backoff.next();
            logger.trace("Retry of bulk request scheduled in {} ms.", next.millis());
            Runnable command = scheduler.preserveContext(() -> this.execute(bulkRequestForRetry));
            retryCancellable = scheduler.schedule(command, next, ThreadPool.Names.SAME);
        }