Use component coder when handling nullable coder in prism. (#34615)

* Invoke component coder in nullable coder.

* Add a check to ensure the correct number of components in other coders.

* Re-enable a previously failed java flatten test.

Author: shunping

runners/prism/java/build.gradle

@@ -110,8 +110,6 @@ def sickbayTests = [
     'org.apache.beam.sdk.transforms.GroupIntoBatchesTest.testWithShardedKeyInGlobalWindow',
 
     // Java side dying during execution.
-    // https://github.com/apache/beam/issues/32930
-    'org.apache.beam.sdk.transforms.FlattenTest.testFlattenMultipleCoders',
     // Stream corruption error java side: failed:java.io.StreamCorruptedException: invalid stream header: 206E6F74
     // Likely due to prism't coder changes.
     'org.apache.beam.sdk.transforms.FlattenTest.testFlattenWithDifferentInputAndOutputCoders2',

sdks/go/pkg/beam/runners/prism/internal/coders.go

@@ -250,6 +250,11 @@ func pullDecoderNoAlloc(c *pipepb.Coder, coders map[string]*pipepb.Coder) func(i
 			ioutilx.ReadN(r, int(l))
 		}
 	case urns.CoderNullable:
+		ccids := c.GetComponentCoderIds()
+		if len(ccids) != 1 {
+			panic(fmt.Sprintf("Nullable coder must have only one component: %s", prototext.Format(c)))
+		}
+		ed := pullDecoderNoAlloc(coders[ccids[0]], coders)
 		return func(r io.Reader) {
 			b, _ := ioutilx.ReadN(r, 1)
 			if len(b) == 0 {
@@ -260,8 +265,7 @@ func pullDecoderNoAlloc(c *pipepb.Coder, coders map[string]*pipepb.Coder) func(i
 			if prefix == 0 {
 				return
 			}
-			l, _ := coder.DecodeVarInt(r)
-			ioutilx.ReadN(r, int(l))
+			ed(r)
 		}
 	case urns.CoderVarInt:
 		return func(r io.Reader) {
@@ -277,14 +281,16 @@ func pullDecoderNoAlloc(c *pipepb.Coder, coders map[string]*pipepb.Coder) func(i
 		}
 	case urns.CoderIterable:
 		ccids := c.GetComponentCoderIds()
+		if len(ccids) != 1 {
+			panic(fmt.Sprintf("Iterable coder must have only one component: %s", prototext.Format(c)))
+		}
 		ed := pullDecoderNoAlloc(coders[ccids[0]], coders)
 		return func(r io.Reader) {
 			l, _ := coder.DecodeInt32(r)
 			for i := int32(0); i < l; i++ {
 				ed(r)
 			}
 		}
-
 	case urns.CoderKV:
 		ccids := c.GetComponentCoderIds()
 		if len(ccids) != 2 {