fix: Read all EventGroups from simulators rather than stopping at first page

[SanjayVas]

The MC and EDP simulators were only reading the first page of results from ListEventGroups. This means that in environments with more EventGroups, not all test EventGroups would be read. This is exacerbated by #1916, as the default page size is now smaller.

[wfa-reviewable]

This change is 

[SanjayVas]

Depends on #1923

[stevenwarejones]

Reviewed 2 of 6 files at r1, 2 of 14 files at r2, all commit messages.
Reviewable status: 4 of 20 files reviewed, 1 unresolved discussion (waiting on @SanjayVas)

---

src/main/kotlin/org/wfanet/measurement/common/api/grpc/ListResources.kt line 67 at r2 (raw file):

      val resourceList: ResourceList<T> = list(nextPageToken, remaining)
      require(resourceList.size <= remaining) {

should this really be required here and throw an error? this is a problem with the list that is passed in..... meaning an API that returned too many items. I'd prefer it took only that many elements if there are too many.

[SanjayVas]

Reviewable status: 4 of 20 files reviewed, 1 unresolved discussion (waiting on @stevenwarejones)

---

src/main/kotlin/org/wfanet/measurement/common/api/grpc/ListResources.kt line 67 at r2 (raw file):

I'd prefer it took only that many elements if there are too many.

That wouldn't work, as the returned next page token wouldn't have the right value to get subsequent items.

[SanjayVas]

Reviewable status: 4 of 20 files reviewed, 1 unresolved discussion (waiting on @stevenwarejones)

---

src/main/kotlin/org/wfanet/measurement/common/api/grpc/ListResources.kt line 67 at r2 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

I'd prefer it took only that many elements if there are too many.

That wouldn't work, as the returned next page token wouldn't have the right value to get subsequent items.

Also GitHub got a bit confused and included changes from the base PR. I assume this comment was meant for #1923

[stevenwarejones]

Reviewed 20 of 20 files at r3, all commit messages.
Reviewable status: all files reviewed, 1 unresolved discussion (waiting on @SanjayVas)

---

src/main/kotlin/org/wfanet/measurement/loadtest/dataprovider/EdpSimulator.kt line 369 at r3 (raw file):

      }
      .flattenConcat()
      .firstOrNull { it.eventGroupReferenceId == eventGroupReferenceId }

I'd prefer you prefilter instead of unncessarily fetching all event groups.... something like

eturn eventGroupsStub
      .listResources(
          limit = 1
) { pageToken ->
        val response =
          try {
            listEventGroups(
              listEventGroupsRequest {
                parent = edpData.name
                filter =
                  ListEventGroupsRequestKt.filter {
                    measurementConsumers += measurementConsumerName
                  }
                this.pageToken = pageToken
              }
            )
          } catch (e: StatusException) {
            throw Exception("Error listing EventGroups", e)
          }
        ResourceList(
response.eventGroupsList.filter { it.eventGroupReferenceId == eventGroupReferenceId }, 
response.nextPageToken
)
      }
      .flattenConcat()
      

[SanjayVas]

Reviewable status: all files reviewed, 1 unresolved discussion (waiting on @stevenwarejones)

---

src/main/kotlin/org/wfanet/measurement/loadtest/dataprovider/EdpSimulator.kt line 369 at r3 (raw file):

Previously, stevenwarejones (Steven Ware Jones) wrote…

I'd prefer you prefilter instead of unncessarily fetching all event groups.... something like

eturn eventGroupsStub
      .listResources(
          limit = 1
) { pageToken ->
        val response =
          try {
            listEventGroups(
              listEventGroupsRequest {
                parent = edpData.name
                filter =
                  ListEventGroupsRequestKt.filter {
                    measurementConsumers += measurementConsumerName
                  }
                this.pageToken = pageToken
              }
            )
          } catch (e: StatusException) {
            throw Exception("Error listing EventGroups", e)
          }
        ResourceList(
response.eventGroupsList.filter { it.eventGroupReferenceId == eventGroupReferenceId }, 
response.nextPageToken
)
      }
      .flattenConcat()

That would be less efficient as it means making a call for one EventGroup at a time rather than batching by the default page size. This will still stop making calls once we get a response that contains the event group.

[SanjayVas]

Reviewable status: all files reviewed, 1 unresolved discussion (waiting on @stevenwarejones)

---

src/main/kotlin/org/wfanet/measurement/loadtest/dataprovider/EdpSimulator.kt line 369 at r3 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

That would be less efficient as it means making a call for one EventGroup at a time rather than batching by the default page size. This will still stop making calls once we get a response that contains the event group.

To clarify, what I mean is that the implementation in this PR will already stop making requests once it gets a response that has an EventGroup that has the specified reference ID. This is because Flow.flattenConcat is an intermediate Flow operator analogous to Sequence.flatten.

Your suggestion does not add any benefit and furthermore does not meet the constraint of never returning more items than limit since EventGroup reference IDs are not yet guaranteed to be unique (see #1913). In order to meet that constraint, it would have to pass the remaining value as the pageSize in the request resulting in the non-batching behavior I mentioned.

[stevenwarejones]

Reviewable status: complete! all files reviewed, all discussions resolved (waiting on @SanjayVas)

---

src/main/kotlin/org/wfanet/measurement/loadtest/dataprovider/EdpSimulator.kt line 369 at r3 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

To clarify, what I mean is that the implementation in this PR will already stop making requests once it gets a response that has an EventGroup that has the specified reference ID. This is because Flow.flattenConcat is an intermediate Flow operator analogous to Sequence.flatten.

Your suggestion does not add any benefit and furthermore does not meet the constraint of never returning more items than limit since EventGroup reference IDs are not yet guaranteed to be unique (see #1913). In order to meet that constraint, it would have to pass the remaining value as the pageSize in the request resulting in the non-batching behavior I mentioned.

I was not aware that Flow.flattenConcat would end up causing the requests to stop once an event group is found.

However, I think you're misunderstanding what what I suggested... I'm suggesting that the listPredicate itself filter out non-matching resource names. That means that it will always return either 1 or 0 items. If it finds no matches, it returns 0 items. Otherwise, it returns 1. Once once is returned, there is no need to continue. I think this is a little more straightforward. The reader may read your code and make the same assumption I did.