Shutdown processing when session expires.

At present the whole loader will shutdown if session expires for even one routine. We are doing this so that we get to know of the problem fast enough as we get Crashloop alerts on continuous restarts. Later we can change the session timeouts to restart the ConsumeClaim and track the error rate prometheus

#150 (comment)

redshiftsink/pkg/redshiftbatcher/batch_processor.go

@@ -177,14 +177,21 @@ func removeEmptyNullValues(value map[string]*string) map[string]*string {
 	return value
 }
 
-// TODO: get rid of this https://github.com/herryg91/gobatch/issues/2
 func (b *batchProcessor) ctxCancelled(ctx context.Context) bool {
 	select {
 	case <-ctx.Done():
 		err := ctx.Err()
-		klog.Warningf("Batch processing stopped, ctx done, ctxErr: %v", err)
-		klog.V(2).Infof(
-			"topic:%s, batchId:%d, lastCommittedOffset:%d: Cancelled.\n",
+		klog.Warningf("Processing stopped! main ctx done, ctxErr: %v", err)
+		klog.Warningf(
+			"%s, batchId:%d, lastCommitted:%d: main ctx done. Cancelled.\n",
+			b.topic, b.batchId, b.lastCommittedOffset,
+		)
+		return true
+	case <-b.session.Context().Done():
+		err := ctx.Err()
+		klog.Warningf("Processing stopped! ctx done, ctxErr: %v", err)
+		klog.Warningf(
+			"%s, batchId:%d, lastCommitted:%d: session ctx done. Cancelled.\n",
 			b.topic, b.batchId, b.lastCommittedOffset,
 		)
 		return true

redshiftsink/pkg/redshiftloader/load_processor.go

@@ -121,12 +121,21 @@ func newLoadProcessor(
 	}
 }
 
-// TODO: get rid of this https://github.com/herryg91/gobatch/issues/2
 func (b *loadProcessor) ctxCancelled(ctx context.Context) bool {
 	select {
 	case <-ctx.Done():
-		klog.Infof(
-			"%s, batchId:%d, lastCommittedOffset:%d: Cancelled.\n",
+		err := ctx.Err()
+		klog.Warningf("Processing stopped! main ctx done, ctxErr: %v", err)
+		klog.Warningf(
+			"%s, batchId:%d, lastCommitted:%d: main ctx done. Cancelled.\n",
+			b.topic, b.batchId, b.lastCommittedOffset,
+		)
+		return true
+	case <-b.session.Context().Done():
+		err := ctx.Err()
+		klog.Warningf("Processing stopped! ctx done, ctxErr: %v", err)
+		klog.Warningf(
+			"%s, batchId:%d, lastCommitted:%d: session ctx done. Cancelled.\n",
 			b.topic, b.batchId, b.lastCommittedOffset,
 		)
 		return true