Handle worker_id is empty string error

sdks/go/pkg/beam/runners/prism/internal/worker/pool.go

@@ -43,6 +43,9 @@ func (m MapW) workerFromMetadataCtx(ctx context.Context) (*W, error) {
 	if err != nil {
 		return nil, err
 	}
+	if id == "" {
+		return nil, fmt.Errorf("worker id in ctx metadata is an empty string")
+	}
 	mu.Lock()
 	defer mu.Unlock()
 	if w, ok := m[id]; ok {

sdks/go/pkg/beam/runners/prism/internal/worker/pool_test.go

@@ -467,6 +467,12 @@ func TestMapW_workerFromMetadataCtx(t *testing.T) {
 			m:       make(MapW),
 			wantErr: "failed to read metadata from context",
 		},
+		{
+			name:    "ctx metadata worker_id=''",
+			ctx:     metadata.NewIncomingContext(context.Background(), metadata.Pairs("worker_id", "")),
+			m:       make(MapW),
+			wantErr: "worker id in ctx metadata is an empty string",
+		},
 		{
 			name: "mismatched ctx metadata",
 			ctx:  metadata.NewIncomingContext(context.Background(), metadata.Pairs("worker_id", "wk1")),

sdks/python/apache_beam/runners/worker/sdk_worker.py

@@ -180,6 +180,8 @@ def __init__(
     # type: (...) -> None
     self._alive = True
     self._worker_index = 0
+    if worker_id == '':
+        raise RuntimeError("worker_id is an empty string")
     self._worker_id = worker_id
     self._state_cache = StateCache(state_cache_size)
     self._deferred_exception = deferred_exception