Repository: redpanda-data/connect (language: go)

Issue:
Input azure_table_storage hangs when the query result is empty
Given a simple pipeline with azure_table_storage as input:

```
input:
  azure_table_storage:
    storage_connection_string: "theConnString"
    table_name: ${TABLE_NAME}
    filter: "PartitionKey eq 'foo'"
    page_size: 1000

```

the pipeline hangs if the result from the query is an empty batch. This is probably the result of a regression introduced in version 4.12.0, the same pipeline works with 4.11.0 and older. 

I am trying to debug the issue but it does not seem to be an obvious one to me. I don't know if this affects other inputs, however, I tried sql_select with postgres and it works correctly.

I will keep trying to debug the issue but let me know if you can help.
Thanks


Discussion / hints:

Fix patch:
diff --git a/internal/impl/azure/input_table_storage.go b/internal/impl/azure/input_table_storage.go
index 66f6de6c70..c67a6edcf6 100644
--- a/internal/impl/azure/input_table_storage.go
+++ b/internal/impl/azure/input_table_storage.go
@@ -132,24 +132,26 @@ func int32OrNil(val int32) *int32 {
 // ReadBatch attempts to read a new page from the target Azure Storage Table.
 func (a *azureTableStorage) ReadBatch(ctx context.Context) (msg message.Batch, ackFn input.AsyncAckFn, err error) {
 	if a.pager.More() {
-		resp, nerr := a.pager.NextPage(ctx)
-		if nerr != nil {
-			a.log.Warnf("error fetching next page", nerr)
+		resp, err := a.pager.NextPage(ctx)
+		if err != nil {
+			a.log.Warnf("error fetching next page", err)
 			return nil, nil, component.ErrTypeClosed
 		}
-		for _, entity := range resp.Entities {
-			p := message.NewPart(entity)
-			msg = append(msg, p)
+		if len(resp.Entities) > 0 {
+			for _, entity := range resp.Entities {
+				p := message.NewPart(entity)
+				msg = append(msg, p)
+			}
+			_ = msg.Iter(func(_ int, part *message.Part) error {
+				a.row++
+				part.MetaSetMut("table_storage_name", a.conf.TableName)
+				part.MetaSetMut("row_num", a.row)
+				return nil
+			})
+			return msg, func(rctx context.Context, res error) error {
+				return nil
+			}, err
 		}
-		_ = msg.Iter(func(_ int, part *message.Part) error {
-			a.row++
-			part.MetaSetMut("table_storage_name", a.conf.TableName)
-			part.MetaSetMut("row_num", a.row)
-			return nil
-		})
-		return msg, func(rctx context.Context, res error) error {
-			return nil
-		}, err
 	}
 	return nil, nil, component.ErrTypeClosed
 }


Test patch: