diff --git a/api/types/state.go b/api/types/state.go
index 46ba3164..d0e2fe9a 100644
--- a/api/types/state.go
+++ b/api/types/state.go
@@ -29,6 +29,9 @@ const (
 	Available                            ConfigurationState = "Available"
 	ConfigurationProvisioningAndChecking ConfigurationState = "ProvisioningAndChecking"
 	ConfigurationDestroying              ConfigurationState = "Destroying"
+	ConfigurationApplyFailed             ConfigurationState = "ApplyFailed"
+	ConfigurationDestroyFailed           ConfigurationState = "DestroyFailed"
+	ConfigurationReloading               ConfigurationState = "ConfigurationReloading"
 )
 
 // ProviderState is the type for Provider state
diff --git a/chart/templates/tf_controller_clusterrole.yml b/chart/templates/tf_controller_clusterrole.yml
index a5571fec..6a5ede81 100644
--- a/chart/templates/tf_controller_clusterrole.yml
+++ b/chart/templates/tf_controller_clusterrole.yml
@@ -29,3 +29,15 @@ rules:
     verbs:
       - "list"
       - "watch"
+
+  - apiGroups:
+      - ""
+    resources:
+      - pods/log
+      - pods
+    verbs:
+      - get
+      - list
+      - create
+      - update
+      - delete
\ No newline at end of file
diff --git a/controllers/configuration/configuration.go b/controllers/configuration/configuration.go
index f915dbab..dfba607b 100644
--- a/controllers/configuration/configuration.go
+++ b/controllers/configuration/configuration.go
@@ -1,12 +1,6 @@
 package configuration
 
 import (
-	"fmt"
-	"os"
-	"os/exec"
-	"path"
-	"strings"
-
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/pkg/errors"
@@ -39,7 +33,7 @@ func ValidConfigurationObject(configuration *v1beta1.Configuration) (types.Confi
 }
 
 // RenderConfiguration will compose the Terraform configuration with hcl/json and backend
-func RenderConfiguration(configuration *v1beta1.Configuration, controllerNamespace string, configurationType types.ConfigurationType, preState bool) (string, error) {
+func RenderConfiguration(configuration *v1beta1.Configuration, controllerNamespace string, configurationType types.ConfigurationType) (string, error) {
 	if configuration.Spec.Backend != nil {
 		if configuration.Spec.Backend.SecretSuffix == "" {
 			configuration.Spec.Backend.SecretSuffix = configuration.Name
@@ -61,9 +55,7 @@ func RenderConfiguration(configuration *v1beta1.Configuration, controllerNamespa
 		return configuration.Spec.JSON, nil
 	case types.ConfigurationHCL:
 		completedConfiguration := configuration.Spec.HCL
-		if !preState {
-			completedConfiguration += "\n" + backendTF
-		}
+		completedConfiguration += "\n" + backendTF
 		return completedConfiguration, nil
 	case types.ConfigurationRemote:
 		return backendTF, nil
@@ -105,62 +97,3 @@ func CompareTwoContainerEnvs(s1 []v1.EnvVar, s2 []v1.EnvVar) bool {
 	}
 	return cmp.Diff(s1, s2, cmpopts.SortSlices(less)) == ""
 }
-
-// checkTerraformSyntax checks the syntax error and state for a HCL/JSON configuration
-func checkTerraformSyntax(name, configuration string) (bool, error) {
-	klog.InfoS("About to check syntax issues", "Configuration", name)
-	state := false
-	dir, osErr := os.MkdirTemp("", fmt.Sprintf("tf-validate-%s-", name))
-	if osErr != nil {
-		klog.ErrorS(osErr, "Failed to create folder", "Dir", dir)
-		return state, osErr
-	}
-	defer os.RemoveAll(dir) //nolint:errcheck
-	tfFile := fmt.Sprintf("%s/main.tf", dir)
-	if err := os.WriteFile(tfFile, []byte(configuration), 0777); err != nil { //nolint
-		klog.ErrorS(err, "Failed to write Configuration hcl to main.tf", "HCL", configuration)
-		return state, err
-	}
-	if err := os.Chdir(dir); err != nil {
-		klog.ErrorS(err, "Failed to change dir", "dir", dir)
-		return state, err
-	}
-
-	var (
-		output []byte
-		err    error
-	)
-	output, err = exec.Command("terraform", "init").CombinedOutput()
-	if err != nil {
-		klog.ErrorS(err, "The command execution isn't successful", "cmd", "terraform init", "output", string(output))
-	} else {
-		output, err = exec.Command("terraform", "validate").CombinedOutput()
-		if err != nil {
-			klog.ErrorS(err, "The command execution isn't successful", "cmd", "terraform validate", "output", string(output))
-		}
-		_, err := os.Stat(path.Join(dir, "/.terraform/terraform.tfstate"))
-		if err == nil {
-			state = true
-		}
-	}
-	if strings.Contains(string(output), "Success!") {
-		return state, nil
-	}
-	return state, errors.New(string(output))
-}
-
-// CheckConfigurationSyntax checks the syntax of Configuration
-func CheckConfigurationSyntax(configuration *v1beta1.Configuration, configurationType types.ConfigurationType) (bool, error) {
-	var template string
-	switch configurationType {
-	case types.ConfigurationHCL:
-		template = configuration.Spec.HCL
-	case types.ConfigurationJSON:
-		template = configuration.Spec.JSON
-	case types.ConfigurationRemote:
-		// TODO(zzxwill) check syntax issue
-		return false, nil
-
-	}
-	return checkTerraformSyntax(configuration.Name, template)
-}
diff --git a/controllers/configuration/configuration_test.go b/controllers/configuration/configuration_test.go
index f4a7250a..49abac4d 100644
--- a/controllers/configuration/configuration_test.go
+++ b/controllers/configuration/configuration_test.go
@@ -1,185 +1 @@
 package configuration
-
-import (
-	"os"
-	"strings"
-	"testing"
-
-	"github.com/google/go-cmp/cmp"
-	"github.com/stretchr/testify/require"
-	v1 "k8s.io/api/core/v1"
-)
-
-func TestCompareTwoContainerEnvs(t *testing.T) {
-	cases := map[string]struct {
-		s1   []v1.EnvVar
-		s2   []v1.EnvVar
-		want bool
-	}{
-		"Equal": {
-			s1: []v1.EnvVar{
-				{
-					Name:  "TF_VAR_zone_id",
-					Value: "cn-beijing-z",
-				},
-				{
-					Name:  "E1",
-					Value: "V1",
-				},
-				{
-					Name:  "NAME",
-					Value: "aaa",
-				},
-			},
-			s2: []v1.EnvVar{
-
-				{
-					Name:  "NAME",
-					Value: "aaa",
-				},
-				{
-					Name:  "TF_VAR_zone_id",
-					Value: "cn-beijing-z",
-				},
-				{
-					Name:  "E1",
-					Value: "V1",
-				},
-			},
-			want: true,
-		},
-		"Not Equal": {
-			s1: []v1.EnvVar{
-				{
-					Name:  "NAME",
-					Value: "aaa",
-				},
-			},
-			s2: []v1.EnvVar{
-
-				{
-					Name:  "NAME",
-					Value: "bbb",
-				},
-			},
-			want: false,
-		},
-	}
-	for name, tc := range cases {
-		t.Run(name, func(t *testing.T) {
-			equal := CompareTwoContainerEnvs(tc.s1, tc.s2)
-			if equal != tc.want {
-				if diff := cmp.Diff(tc.want, equal); diff != "" {
-					t.Errorf("\nCompareTwoContainerEnvs(...) %s\n", diff)
-				}
-			}
-		})
-	}
-}
-
-func TestCheckTFConfiguration(t *testing.T) {
-	cases := map[string]struct {
-		name          string
-		configuration string
-		subStr        string
-		state         bool
-	}{
-		"Invalid": {
-			name: "bad",
-			configuration: `resource2 "alicloud_oss_bucket" "bucket-acl" {
-  bucket = var.bucket
-  acl = var.acl
-}
-
-output "BUCKET_NAME" {
-  value = "${alicloud_oss_bucket.bucket-acl.bucket}.${alicloud_oss_bucket.bucket-acl.extranet_endpoint}"
-}
-
-variable "bucket" {
-  description = "OSS bucket name"
-  default = "vela-website"
-  type = string
-}
-
-variable "acl" {
-  description = "OSS bucket ACL, supported 'private', 'public-read', 'public-read-write'"
-  default = "private"
-  type = string
-}
-`,
-			subStr: "Error:",
-			state:  false,
-		},
-		"valid": {
-			name: "good",
-			configuration: `resource "alicloud_oss_bucket" "bucket-acl" {
-  bucket = var.bucket
-  acl = var.acl
-}
-
-output "BUCKET_NAME" {
-  value = "${alicloud_oss_bucket.bucket-acl.bucket}.${alicloud_oss_bucket.bucket-acl.extranet_endpoint}"
-}
-
-variable "bucket" {
-  description = "OSS bucket name"
-  default = "vela-website"
-  type = string
-}
-
-variable "acl" {
-  description = "OSS bucket ACL, supported 'private', 'public-read', 'public-read-write'"
-  default = "private"
-  type = string
-}`,
-			subStr: "",
-			state:  false,
-		},
-		"valid-with-state": {
-			name: "good",
-			configuration: `resource "alicloud_oss_bucket" "bucket-acl" {
-  bucket = var.bucket
-  acl = var.acl
-}
-
-output "BUCKET_NAME" {
-  value = "${alicloud_oss_bucket.bucket-acl.bucket}.${alicloud_oss_bucket.bucket-acl.extranet_endpoint}"
-}
-
-variable "bucket" {
-  description = "OSS bucket name"
-  default = "vela-website"
-  type = string
-}
-
-variable "acl" {
-  description = "OSS bucket ACL, supported 'private', 'public-read', 'public-read-write'"
-  default = "private"
-  type = string
-}
-
-terraform {
-  backend "local" {
-    path = "./test.tfstate"
-  }
-}`,
-			subStr: "",
-			state:  true,
-		},
-	}
-	// As the entry point is the root folder `terraform-controller`, the unit-test locates here `./controllers/configuration`,
-	// so we change the directory
-	os.Chdir("../../")
-	for name, tc := range cases {
-		t.Run(name, func(t *testing.T) {
-			r := require.New(t)
-			state, err := checkTerraformSyntax(tc.name, tc.configuration)
-			if err != nil {
-				if !strings.Contains(err.Error(), tc.subStr) {
-					t.Errorf("\ncheckTFConfiguration(...) %s\n", cmp.Diff(err.Error(), tc.subStr))
-				}
-			}
-			r.Equal(state, tc.state)
-		})
-	}
-}
diff --git a/controllers/configuration_controller.go b/controllers/configuration_controller.go
index 4a8d8837..40723df6 100644
--- a/controllers/configuration_controller.go
+++ b/controllers/configuration_controller.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"math"
 	"os"
 	"time"
 
@@ -33,13 +34,14 @@ import (
 	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	"github.com/oam-dev/terraform-controller/api/types"
 	crossplane "github.com/oam-dev/terraform-controller/api/types/crossplane-runtime"
 	"github.com/oam-dev/terraform-controller/api/v1beta1"
 	cfgvalidator "github.com/oam-dev/terraform-controller/controllers/configuration"
+	"github.com/oam-dev/terraform-controller/controllers/terraform"
 	"github.com/oam-dev/terraform-controller/controllers/util"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
 
 const (
@@ -99,6 +101,8 @@ const (
 	ErrProviderNotReady = "Provider is not ready"
 	// MessageProviderReady means provider object is ready
 	MessageProviderReady = "Provider is ready"
+	// ConfigurationReloading means Configuration changed and needs reloading
+	ConfigurationReloading = "Configuration has changed and is reloading"
 )
 
 // ConfigurationReconciler reconciles a Configuration object.
@@ -163,6 +167,7 @@ func (r *ConfigurationReconciler) Reconcile(req ctrl.Request) (ctrl.Result, erro
 		}
 	}
 
+	// add finalizer
 	if configuration.ObjectMeta.DeletionTimestamp.IsZero() {
 		if !controllerutil.ContainsFinalizer(&configuration, configurationFinalizer) {
 			controllerutil.AddFinalizer(&configuration, configurationFinalizer)
@@ -172,13 +177,22 @@ func (r *ConfigurationReconciler) Reconcile(req ctrl.Request) (ctrl.Result, erro
 		}
 	}
 
+	// pre-check Configuration
 	if err := r.preCheck(ctx, &configuration, meta); err != nil {
 		return ctrl.Result{}, err
 	}
 
 	if !configuration.ObjectMeta.DeletionTimestamp.IsZero() {
 		// terraform destroy
-		klog.InfoS("Performing Configuration Destroy", "Namespace", req.Namespace, "Name", req.Name, "JobName", meta.DestroyJobName)
+		klog.InfoS("performing Configuration Destroy", "Namespace", req.Namespace, "Name", req.Name, "JobName", meta.DestroyJobName)
+
+		if err := terraform.GetTerraformStatus(ctx, meta.Namespace, meta.DestroyJobName); err != nil {
+			klog.ErrorS(err, "Terraform destroy failed")
+			if updateErr := updateStatus(ctx, r.Client, configuration, types.ConfigurationDestroyFailed, err.Error()); updateErr != nil {
+				return ctrl.Result{}, err
+			}
+		}
+
 		if err := r.terraformDestroy(ctx, configuration, meta); err != nil {
 			if err.Error() == MessageDestroyJobNotCompleted {
 				return ctrl.Result{RequeueAfter: 3 * time.Second}, nil
@@ -195,13 +209,16 @@ func (r *ConfigurationReconciler) Reconcile(req ctrl.Request) (ctrl.Result, erro
 	}
 
 	// Terraform apply (create or update)
-	if configuration.Status.Apply.State == types.ConfigurationSyntaxError {
-		return ctrl.Result{RequeueAfter: 3 * time.Second}, nil
-	}
 	klog.InfoS("performing Terraform Apply (cloud resource create/update)", "Namespace", req.Namespace, "Name", req.Name)
 	if configuration.Spec.ProviderReference != nil {
 		r.ProviderName = configuration.Spec.ProviderReference.Name
 	}
+	if err := terraform.GetTerraformStatus(ctx, meta.Namespace, meta.ApplyJobName); err != nil {
+		klog.ErrorS(err, "Terraform apply failed")
+		if updateErr := updateStatus(ctx, r.Client, configuration, types.ConfigurationApplyFailed, err.Error()); updateErr != nil {
+			return ctrl.Result{}, err
+		}
+	}
 	if err := r.terraformApply(ctx, req.Namespace, configuration, meta); err != nil {
 		if err.Error() == MessageApplyJobNotCompleted {
 			return ctrl.Result{RequeueAfter: 3 * time.Second}, nil
@@ -220,10 +237,10 @@ func (r *ConfigurationReconciler) terraformApply(ctx context.Context, namespace
 	)
 
 	// start provisioning and check the status of the provision
-	if configuration.Status.Apply.State != types.Available && configuration.Status.Apply.State != types.ProviderNotReady {
-		configuration.Status.Apply = v1beta1.ConfigurationApplyStatus{
-			State:   types.ConfigurationProvisioningAndChecking,
-			Message: MessageCloudResourceProvisioningAndChecking,
+	if configuration.Status.Apply.State != types.Available && configuration.Status.Apply.State != types.ProviderNotReady &&
+		configuration.Status.Apply.State != types.ConfigurationApplyFailed {
+		if err := updateStatus(ctx, k8sClient, configuration, types.ConfigurationProvisioningAndChecking, MessageCloudResourceProvisioningAndChecking); err != nil {
+			return err
 		}
 	}
 
@@ -239,12 +256,11 @@ func (r *ConfigurationReconciler) terraformApply(ctx context.Context, namespace
 	}
 
 	if tfExecutionJob.Status.Succeeded == int32(1) && configuration.Status.Apply.State != types.Available {
-		configuration.Status.Apply = v1beta1.ConfigurationApplyStatus{
-			State:   types.Available,
-			Message: MessageCloudResourceDeployed,
+		if err := updateStatus(ctx, k8sClient, configuration, types.Available, MessageCloudResourceDeployed); err != nil {
+			return err
 		}
 	}
-	return updateStatus(ctx, k8sClient, configuration, configuration.Status.Apply.State, configuration.Status.Apply.Message)
+	return nil
 }
 
 func (r *ConfigurationReconciler) terraformDestroy(ctx context.Context, configuration v1beta1.Configuration, meta *TFConfigurationMeta) error {
@@ -321,17 +337,10 @@ func (r *ConfigurationReconciler) preCheck(ctx context.Context, configuration *v
 	}
 	meta.ConfigurationType = configurationType
 
-	// Validation: 2) validate Configuration syntax
-	preState, err := cfgvalidator.CheckConfigurationSyntax(configuration, configurationType)
-	if err != nil {
-		return updateStatus(ctx, k8sClient, *configuration, types.ConfigurationSyntaxError, err.Error())
-	}
-	if configuration.Status.Apply.State == types.ConfigurationSyntaxError {
-		updateStatus(ctx, k8sClient, *configuration, types.ConfigurationSyntaxGood, "") //nolint:errcheck
-	}
+	// TODO(zzxwill) Need to find an alternative to check whether there is an state backend in the Configuration
 
 	// Render configuration with backend
-	completeConfiguration, err := cfgvalidator.RenderConfiguration(configuration, controllerNamespace, configurationType, preState)
+	completeConfiguration, err := cfgvalidator.RenderConfiguration(configuration, controllerNamespace, configurationType)
 	if err != nil {
 		return err
 	}
@@ -351,8 +360,12 @@ func (r *ConfigurationReconciler) preCheck(ctx context.Context, configuration *v
 	if err != nil {
 		return err
 	}
+
 	meta.ConfigurationChanged = configurationChanged
 	if configurationChanged {
+		if err := updateStatus(ctx, k8sClient, *configuration, types.ConfigurationReloading, ConfigurationReloading); err != nil {
+			return err
+		}
 		// store configuration to ConfigMap
 		return meta.storeTFConfiguration(ctx, k8sClient)
 	}
@@ -430,6 +443,7 @@ func (meta *TFConfigurationMeta) assembleTerraformJob(executionType TerraformExe
 		initContainers []v1.Container
 		parallelism    int32 = 1
 		completions    int32 = 1
+		backoffLimit   int32 = math.MaxInt32
 	)
 
 	executorVolumes := meta.assembleExecutorVolumes()
@@ -487,8 +501,9 @@ func (meta *TFConfigurationMeta) assembleTerraformJob(executionType TerraformExe
 			Namespace: controllerNamespace,
 		},
 		Spec: batchv1.JobSpec{
-			Parallelism: &parallelism,
-			Completions: &completions,
+			Parallelism:  &parallelism,
+			Completions:  &completions,
+			BackoffLimit: &backoffLimit,
 			Template: v1.PodTemplateSpec{
 				Spec: v1.PodSpec{
 					// InitContainer will copy Terraform configuration files to working directory and create Terraform
diff --git a/controllers/terraform/logging.go b/controllers/terraform/logging.go
new file mode 100644
index 00000000..da435acd
--- /dev/null
+++ b/controllers/terraform/logging.go
@@ -0,0 +1,53 @@
+package terraform
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/klog/v2"
+)
+
+func initClientSet() (*kubernetes.Clientset, error) {
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, err
+	}
+	return kubernetes.NewForConfig(config)
+}
+
+func getPodLog(ctx context.Context, client *kubernetes.Clientset, namespace, jobName string) (string, error) {
+	label := fmt.Sprintf("job-name=%s", jobName)
+	pods, err := client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: label})
+	if err != nil || pods == nil || len(pods.Items) == 0 {
+		klog.InfoS("pods are not found", "Label", label)
+		return "", nil //nolint:nilerr
+	}
+	pod := pods.Items[0]
+
+	req := client.CoreV1().Pods(namespace).GetLogs(pod.Name, &v1.PodLogOptions{})
+	logs, err := req.Stream(ctx)
+	if err != nil {
+		return "", err
+	}
+	defer func(logs io.ReadCloser) {
+		err := logs.Close()
+		if err != nil {
+			return
+		}
+	}(logs)
+
+	var buf = &bytes.Buffer{}
+	_, err = io.Copy(buf, logs)
+	if err != nil {
+		return "", err
+	}
+	logContent := buf.String()
+	klog.Info("pod logs", "Pod", pod.Name, "Logs", logContent)
+	return logContent, nil
+}
diff --git a/controllers/terraform/status.go b/controllers/terraform/status.go
new file mode 100644
index 00000000..77c10f1a
--- /dev/null
+++ b/controllers/terraform/status.go
@@ -0,0 +1,43 @@
+package terraform
+
+import (
+	"context"
+	"strings"
+
+	"github.com/pkg/errors"
+	"k8s.io/klog/v2"
+)
+
+// GetTerraformStatus will get Terraform execution status
+func GetTerraformStatus(ctx context.Context, namespace, jobName string) error {
+	klog.InfoS("checking Terraform execution status", "Namespace", namespace, "Job", jobName)
+	clientSet, err := initClientSet()
+	if err != nil {
+		klog.ErrorS(err, "failed to init clientSet")
+		return err
+	}
+
+	logs, err := getPodLog(ctx, clientSet, namespace, jobName)
+	if err != nil {
+		klog.ErrorS(err, "failed to get pod logs")
+		return err
+	}
+
+	success, errMsg := analyzeTerraformLog(logs)
+	if success {
+		return nil
+	}
+
+	return errors.New(errMsg)
+}
+
+func analyzeTerraformLog(logs string) (bool, string) {
+	lines := strings.Split(logs, "\n")
+	for i, line := range lines {
+		if strings.Contains(line, "31mError:") {
+			errMsg := strings.Join(lines[i:], "\n")
+			return false, errMsg
+		}
+	}
+	return true, ""
+}
diff --git a/go.mod b/go.mod
index c5259d93..78a360fa 100644
--- a/go.mod
+++ b/go.mod
@@ -12,7 +12,6 @@ require (
 	github.com/onsi/ginkgo v1.16.2
 	github.com/onsi/gomega v1.12.0
 	github.com/pkg/errors v0.9.1
-	github.com/stretchr/testify v1.7.0
 	google.golang.org/appengine v1.6.5 // indirect
 	k8s.io/api v0.18.8
 	k8s.io/apimachinery v0.18.8