Warn for parameterized collection ValueState types

ValueState<Set<T>>, ValueState<List<T>> and ValueState<Map<K, V>> were
silently skipped because the resolved value type has unresolved parameters
(the element type is a type variable). The collection's own raw type is
known in these cases, so the specialized-state recommendation (SetState /
BagState or OrderedListState / MapState) can still be made. Drop the
blanket hasUnresolvedParameters() early-return and match on the raw
collection type instead; a payload that is itself a bare type variable
(ValueState<T>) still produces no recommendation.

Add unit tests covering the parameterized Set/List cases and the bare
type-variable no-warning case.

Author: PDGGK

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java

@@ -2355,12 +2355,11 @@ private static void warnIfValueStateContainsCollection(
     TypeDescriptor<?> valueTypeDescriptor =
         stateType.resolveType(ValueState.class.getTypeParameters()[0]);
 
-    // Skip if the type has unresolved parameters (e.g., TypeVariable, WildcardType)
-    if (valueTypeDescriptor.hasUnresolvedParameters()) {
-      return;
-    }
-
-    // Use TypeDescriptor.isSubtypeOf() for type checking - stays in TypeDescriptor API
+    // Match on the collection's raw type. We intentionally do not skip types with unresolved
+    // parameters: for a parameterized collection such as ValueState<Set<T>> the collection's own
+    // raw type (Set) is known even though the element type T is a type variable, so we can still
+    // recommend SetState. A payload that is itself a bare type variable (e.g. ValueState<T>) simply
+    // matches none of the branches below and produces no recommendation.
     String recommendation = null;
     if (valueTypeDescriptor.isSubtypeOf(TypeDescriptor.of(Map.class))) {
       recommendation = "MapState";

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/reflect/DoFnSignaturesTest.java

@@ -51,6 +51,7 @@
 import org.apache.beam.sdk.state.TimerSpecs;
 import org.apache.beam.sdk.state.ValueState;
 import org.apache.beam.sdk.state.WatermarkHoldState;
+import org.apache.beam.sdk.testing.ExpectedLogs;
 import org.apache.beam.sdk.testing.SerializableMatchers;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.Sum;
@@ -101,6 +102,8 @@ public class DoFnSignaturesTest {
 
   @Rule public ExpectedException thrown = ExpectedException.none();
 
+  @Rule public ExpectedLogs expectedLogs = ExpectedLogs.none(DoFnSignatures.class);
+
   @Test
   public void testBasicDoFnProcessContext() throws Exception {
     DoFnSignature sig =
@@ -1735,6 +1738,32 @@ private static class DoFnWithSimpleValueState extends DoFn<String, String> {
     public void process() {}
   }
 
+  private static class DoFnWithParameterizedSetValueState<T> extends DoFn<String, String> {
+    @StateId("parameterizedSetState")
+    private final StateSpec<ValueState<java.util.Set<T>>> parameterizedSetState =
+        StateSpecs.value();
+
+    @ProcessElement
+    public void process() {}
+  }
+
+  private static class DoFnWithParameterizedListValueState<T> extends DoFn<String, String> {
+    @StateId("parameterizedListState")
+    private final StateSpec<ValueState<java.util.List<T>>> parameterizedListState =
+        StateSpecs.value();
+
+    @ProcessElement
+    public void process() {}
+  }
+
+  private static class DoFnWithTypeVariableValueState<T> extends DoFn<String, String> {
+    @StateId("typeVariableState")
+    private final StateSpec<ValueState<T>> typeVariableState = StateSpecs.value();
+
+    @ProcessElement
+    public void process() {}
+  }
+
   @Test
   public void testValueStateWithMapLogsWarning() {
     // This test verifies that the signature can be parsed for DoFns with collection ValueState.
@@ -1760,5 +1789,32 @@ public void testValueStateWithSimpleTypeNoWarning() {
     // Simple types should not trigger any warning
     DoFnSignature signature = DoFnSignatures.getSignature(DoFnWithSimpleValueState.class);
     assertThat(signature.stateDeclarations().get("simpleState"), notNullValue());
+    expectedLogs.verifyNotLogged("with collection type");
+  }
+
+  @Test
+  public void testValueStateWithParameterizedSetLogsWarning() {
+    // A parameterized collection such as ValueState<Set<T>> still has a known raw collection type
+    // (Set) even though the element type is a type variable, so it should still recommend SetState.
+    DoFnSignature signature = DoFnSignatures.getSignature(DoFnWithParameterizedSetValueState.class);
+    assertThat(signature.stateDeclarations().get("parameterizedSetState"), notNullValue());
+    expectedLogs.verifyWarn("SetState");
+  }
+
+  @Test
+  public void testValueStateWithParameterizedListLogsWarning() {
+    DoFnSignature signature =
+        DoFnSignatures.getSignature(DoFnWithParameterizedListValueState.class);
+    assertThat(signature.stateDeclarations().get("parameterizedListState"), notNullValue());
+    expectedLogs.verifyWarn("BagState or OrderedListState");
+  }
+
+  @Test
+  public void testValueStateWithBareTypeVariableNoWarning() {
+    // A bare type variable payload (ValueState<T>) has no raw collection type to inspect and must
+    // not produce a collection recommendation.
+    DoFnSignature signature = DoFnSignatures.getSignature(DoFnWithTypeVariableValueState.class);
+    assertThat(signature.stateDeclarations().get("typeVariableState"), notNullValue());
+    expectedLogs.verifyNotLogged("with collection type");
   }
 }