Issue 1:
Apache Airflow version: 1.7.1.2

Kubernetes version (if you are using kubernetes) (use kubectl version):

Environment:

Cloud provider or hardware configuration:

OS (e.g. from /etc/os-release):

Kernel (e.g. uname -a):

Install tools:

Others:

What happened:

#1514 added a verify_integrity function that greedily creates TaskInstance objects for all tasks in a dag.

This does not interact well with the assumptions in the new update_state function. The guard for if len(tis) == len(dag.active_tasks) is no longer effective; in the old world of lazily-created tasks this code would only run once all the tasks in the dag had run. Now it runs all the time, and as soon as one task in a dag run fails the whole DagRun fails. This is bad since the scheduler stops processing the DagRun after that.

In retrospect, the old code was also buggy: if your dag ends with a bunch of Queued tasks the DagRun could be marked as failed prematurely.

I suspect the fix is to update the guard to look at tasks where the state is success or failed. Otherwise we're evaluating and failing the dag based on up_for_retry/queued/scheduled tasks.

What you expected to happen:

How to reproduce it:

Anything else we need to know:

Moved here from https://issues.apache.org/jira/browse/AIRFLOW-441

Issue 2:
Body

Provider alibaba: 1.1.0rc1

Add oss_task_handler into alibaba-provider and enable remote logging to OSS (#21785): @EricGao888

Provider amazon: 3.1.1rc1

Added AWS RDS sensors (#21231): @kazanzhy

Added AWS RDS operators (#20907): @kazanzhy

Add RedshiftDataHook (#19137): @john-jac

Feature: Add invoke lambda function operator (#21686): @schirag1993

Add JSON output on SqlToS3Operator (#21779): @mariotaddeucci

Implement a Sagemaker DeleteModelOperator and Delete model hook. (#21673): @hsrocks

Added Hook for Amazon RDS (#20642): @kazanzhy

Added SNS example DAG and rst (#21475): @ferruzzi

retry on very specific eni provision failures (#22002): @zachliu

Configurable AWS Session Factory (#21778): @ac1997

S3KeySensor to use S3Hook url parser (#21500): @dstandish

ECSOperator: Get log events after sleep to get all logs (#21574): @kanga333

Use temporary file in GCSToS3Operator (#21295): @rafalh

AWS RDS integration fixes (#22125): @kazanzhy

Fix the Type Hints in RedshiftSQLOperator (#21885): @kaxil

Bug Fix - S3DeleteObjectsOperator will try and delete all keys (#21458): @njrs92

Fix Amazon SES emailer signature (#21681): @potiuk

Fix EcsOperatorError, so it can be loaded from a picklefile (#21441): @ngwallace

Fix RedshiftDataOperator and update doc (#22157): @vincbeck

bugfix (#22137): @zachliu

If uploading task logs to S3 fails, retry once (#21981): @steveyz-astro

Bug-fix GCSToS3Operator (#22071): @rsg17

Refactor query status polling logic in EMRContainerHook (#21423): @victorphoenix3

use different logger to avoid duplicate log entry (#22256): @zachliu

[doc] Improve s3 operator example by adding task upload_keys (#21422): @zhongjiajie

Rename 'S3' hook name to 'Amazon S3' (#21988): @treyyi

Add template fields to DynamoDBToS3Operator (#22080): @eladkal

Provider databricks: 2.4.0rc1

Add new options to DatabricksCopyIntoOperator (#22076): @alexott

Databricks hook - retry on HTTP Status 429 as well (#21852): @alexott

Skip some tests for Databricks from running on Python 3.10 (#22221): @potiuk

Provider docker: 2.5.1rc1

Avoid trying to kill container when it did not succeed for Docker (#22145): @potiuk

Provider google: 6.6.0rc1

Support Uploading Bigger Files to Google Drive (#22179): @ulsc

Make chunk_size in GoogleDriveHook standard (#22222): @ulsc

Add guide for DataprocInstantiateInlineWorkflowTemplateOperator (#22062): @NiloFreitas

Allow GCS Metadata to be included in GCS Upload (#22058): @patricker

Dataplex operators (#20377): @wojsamjan

Add support for ARM platform (#22127): @potiuk

Use yaml safe load (#22091): @lwyszomi

Provider snowflake: 2.5.2rc1

Remove Snowflake limits (#22181): @potiuk

The guidelines on how to test providers can be found in

Verify providers by contributors

Committer

I acknowledge that I am a maintainer/committer of the Apache Airflow project.