Release/0.7.8

.github/workflows/cd.yml

@@ -10,7 +10,7 @@ jobs:
     name: Release
     strategy:
       matrix:
-        go-version: [1.24.11]
+        go-version: [1.25.11]
         os: [ubuntu-latest]
     runs-on: ${{ matrix.os }}
 

.github/workflows/ci.yml

@@ -12,7 +12,7 @@ jobs:
     name: Compile & Test
     strategy:
       matrix:
-        go-version: [1.24.11]
+        go-version: [1.25.11]
         os: [ubuntu-latest]
     runs-on: ${{ matrix.os }}
 

CHANGELOG

@@ -1,3 +1,8 @@
+Version 0.7.8 (2026-06-08)
+--------------------------
+Bump Go toolchain to 1.25.11 (#101)
+Log EMR cluster StateChangeReason on transient mode failure (#100)
+
 Version 0.7.7 (2026-05-22)
 --------------------------
 Log EMR cluster StateChangeReason on failure (#98)

README.md

@@ -21,7 +21,7 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
 
-[release-image]: https://img.shields.io/badge/release-0.7.7-6ad7e5.svg?style=flat
+[release-image]: https://img.shields.io/badge/release-0.7.8-6ad7e5.svg?style=flat
 [releases]: https://github.com/snowplow/dataflow-runner/releases
 
 [license-image]: https://img.shields.io/badge/license-Apache--2-blue.svg?style=flat

VERSION

@@ -1 +1 @@
-0.7.7
+0.7.8

go.mod

@@ -1,6 +1,6 @@
 module github.com/snowplow/dataflow-runner
 
-go 1.24.11
+go 1.25.11
 
 require (
 	github.com/aws/aws-sdk-go-v2 v1.41.5
@@ -60,7 +60,7 @@ require (
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/snowplow-devops/go-retry v0.0.0-20210106090855-8989bbdbae1c
-	golang.org/x/sys v0.40.0 // indirect
+	golang.org/x/sys v0.44.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -268,8 +268,8 @@ golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220728004956-3c1f35247d10/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ=
-golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
+golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ=
+golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=

src/emr_cluster.go

@@ -182,6 +182,20 @@ func clusterStateChangeReason(status *types.ClusterStatus) (string, string) {
 	return code, msg
 }
 
+// fetchStateChangeReason describes the cluster and returns its StateChangeReason code and message.
+// It returns a non-nil error if the cluster cannot be described, or empty strings if no reason is set.
+func (ec EmrCluster) fetchStateChangeReason(ctx context.Context, jobflowID string) (string, string, error) {
+	resp, err := ec.Svc.DescribeCluster(ctx, &emr.DescribeClusterInput{ClusterId: aws.String(jobflowID)})
+	if err != nil {
+		return "", "", err
+	}
+	if resp.Cluster == nil || resp.Cluster.Status == nil {
+		return "", "", nil
+	}
+	code, message := clusterStateChangeReason(resp.Cluster.Status)
+	return code, message, nil
+}
+
 // waitForClusterReady waits for the cluster to reach RUNNING or WAITING state using SDK v2 waiter.
 // Returns the cluster status even on failure (needed for bootstrap failure detection).
 func (ec EmrCluster) waitForClusterReady(ctx context.Context, jobflowID string) (*types.ClusterStatus, error) {

src/emr_cluster_test.go

@@ -228,6 +228,29 @@ func TestRunJobFlow_Success(t *testing.T) {
 	assert.Equal(t, "j-WAITING", id)
 }
 
+func TestFetchStateChangeReason(t *testing.T) {
+	assert := assert.New(t)
+	ctx := context.Background()
+	record, _ := CR.ParseClusterRecord([]byte(ClusterRecord1), nil, "")
+	ec := mockEmrCluster(*record)
+
+	// A cluster with a StateChangeReason returns its code and message
+	code, message, err := ec.fetchStateChangeReason(ctx, "j-TERMINATED_WITH_ERRORS")
+	assert.Nil(err)
+	assert.Equal("VALIDATION_ERROR", code)
+	assert.Equal("On the master instance, application provisioning failed", message)
+
+	// A cluster with no StateChangeReason returns empty strings and no error
+	code, message, err = ec.fetchStateChangeReason(ctx, "j-STARTING")
+	assert.Nil(err)
+	assert.Equal("", code)
+	assert.Equal("", message)
+
+	// A DescribeCluster failure is returned as an error
+	_, _, err = ec.fetchStateChangeReason(ctx, "bad")
+	assert.NotNil(err)
+}
+
 func TestGetJobFlowInput_Success(t *testing.T) {
 	assert := assert.New(t)
 

src/main.go

@@ -34,7 +34,7 @@ const (
 	appName          = "dataflow-runner"
 	appUsage         = "Run templatable playbooks of Hadoop/Spark/et al jobs on Amazon EMR"
 	appCopyright     = "(c) 2016-2026 Snowplow Analytics Ltd"
-	cliVersion       = "0.7.7"
+	cliVersion       = "0.7.8"
 	varDelim         = ","
 	fEmrConfig       = "emr-config"
 	fEmrPlaybook     = "emr-playbook"
@@ -327,6 +327,15 @@ func main() {
 					maxClusterWaitDuration,
 				)
 				if err != nil {
+					// The terminated waiter only returns a generic "transitioned to Failure" error, so
+					// re-describe the cluster to recover and surface the underlying StateChangeReason.
+					if code, message, derr := emrCluster.fetchStateChangeReason(context.Background(), jobFlowSteps.JobflowID); derr != nil {
+						log.Warnf("could not re-describe cluster to get StateChangeReason: %v", derr)
+					} else if code != "" || message != "" {
+						log.Errorf("EMR cluster state change reason: code='%s' message=%q", code, message)
+						err = fmt.Errorf("EMR cluster %s terminated with errors: code=%s message=%q",
+							jobFlowSteps.JobflowID, code, message)
+					}
 					if lock != nil && softLock != "" {
 						lock.Unlock()
 					}