Flink - Fix flakey, order-dependent FlinkTableSource tests for Flink 1.14

[kbendick]

When upgrading our Flink code for the upcoming Flink 1.15 release in #4553, it was discovered that there is a failing test: #4553 (comment)

Looking closer, the expected and actual results are the same, just in a different order.

Because we are not testing the ordering of records returned from the source here, using list comparison doesn't make sense and we can just use set comparison (with an initial size check) for any query that returns more than 1 record.

I have updated all queries in this test to use set comparison semantics as list comparison is inherently flakey.

I will port this over to Flink 1.15 in #4553 as that's where tests are failing (though we don't have any guarantee that they will pass at present in 1.14 and it's simply implementation specific that they do).

cc @openinx @yittg

[danielcweeks]

This is one of those cases where AssertJ could help with convenience checks like assertThat(actual).hasSameElementsAs(expected), but since we're not using that library here yet, this seems like a reasonable approach. Alternatively, you could do something like the following wrapped in a utility function:

assertTrue(expected.size() == actual.size() && expected.containsAll(actual) && actual.containsAll(expected));
``

[kbendick]

Yeah adding a convenience function makes sense.

[kbendick]

Actually, it seems we do have the AssertJ library available for this subproject, so I'll just use that unless there's opposition to it.

[kbendick]

Added a utility function assertSameElements into FlinkTestBase.

In the interest of keeping the diff minimal, I didn't remove existing assertions on the size of the result (although that is checked by assertSameElements).

I also didn't update any of the results with only one element. If we want to do that, we can use Collections.singleton to reuse the comparison.

Let me know if removing the (redundant) size checks is a good idea. I think so because in the case of a failure, we'll see the incorrect records instead of just the incorrect size but I opted to be conservative and change less code.

[yittg]

How about keeping this description by Assertions.assertThat(...).as(...)

[kbendick]

For AssertJ, the assertion message is already pretty good. I updated one test so it would fail and here's the output I got. I think it's pretty clean / self explanatory.

java.lang.AssertionError: 
Expecting:
  [+I[1, iceberg, 10.0], +I[2, b, 20.0]]
to contain exactly in any order:
  [+I[1, iceberg, 10.0]]
but the following elements were unexpected:
  [+I[2, b, 20.0]]

With .as before the containsExactlyInAnyOrderElementsOf, here's the output. I don't personally think it provides much value.

java.lang.AssertionError: [should contain the same elements] 
Expecting:
  [+I[1, iceberg, 10.0], +I[2, b, 20.0]]
to contain exactly in any order:
  [+I[1, iceberg, 10.0]]
but the following elements were unexpected:
  [+I[2, b, 20.0]]

But I'm happy to add it if you think it does.

[yittg]

I think we should make the assertSameElements open(can be optional) for description.
For the case you mentioned, i think it didn't provide much value because the statement itself, not for all other descriptions.
WDYT @kbendick ?

[kbendick]

So you mean add a description argument like assertSameElements(String errorMessage, List<Row> exepected, List<Row> actual)?

I'm happy to add that as well as keep the original signature. I can't think of a place in the current file that it would be useful as they're all the same scenarios, but I'm happy to add it.

[kbendick]

Added that signature in addition to the current one should anybody want to use it @yittg.

I can see how it would provide some benefit in some situations.

[yittg]

thanks @kbendick