[connector] Fluss support scan.startup.timestamp is larger than max timestamp of bucket. #284
Conversation
| @@ -166,7 +166,7 @@ void testPkTableReadOnlySnapshot() throws Exception { | |||
|
|
|||
| @Test | |||
| void testNonPkTableRead() throws Exception { | |||
| tEnv.executeSql("create table non_pk_table_test (a int, b varchar)"); | |||
| tEnv.executeSql("create table f (a int, b varchar)"); | |||
| @@ -631,6 +631,49 @@ void testReadPrimaryKeyPartitionedTable() throws Exception { | |||
| assertResultsIgnoreOrder(rowIter, expectedRowValues, true); | |||
| } | |||
|
|
|||
| @Test | |||
| void testReadTimestampOutOfBounds() throws Exception { | |||
|
|
|||
| ApiException exception = | ||
| ApiError.fromErrorMessage(resp).exception(); | ||
| if (exception | ||
| instanceof InvalidTimestampException) { |
There was a problem hiding this comment.
It's not encouraged to use exceptions as common information exchanging. I would recommend this can be handled in server-side to simplify client-side implementation (and future other clients).
| // requirement of the timestamp have not been produced to fluss yet. In this case, we just | ||
| // use the latest offset. We need to get the latest offsets before querying offsets by time | ||
| // to ensure that no message is going to be missed. | ||
| Map<Integer, Long> endOffsets = latestOffsets(partitionName, buckets); |
There was a problem hiding this comment.
I recommend this fallback logic should be handled in server-side to have a data consistency and simplify client.
From the perspective of ListOffset(timestamp) RPC method semantics, it should "return the offset of the first message whose timestamp is greater than or equals to the given timestamp. If no such message is found, the log end offset is returned." We can still throw InvalidTimestampException if the timestamp is larger than current timestamp (a future time).
And then, adds unit tests in ReplicaManagerTest#testListOffsetsWithTimestamp.
This is also how kafka implement it: https://github.com/apache/kafka/blob/b40a962a8ead04b10f7a8eba346fe79784de9e4e/core/src/main/scala/kafka/log/UnifiedLog.scala#L1252-L1256
loserwang1024
force-pushed
the
timestamp_outbound
branch
from
00fa8b7 to
1ae71b4
Compare
loserwang1024
force-pushed
the
timestamp_outbound
branch
from
1ae71b4 to
31e52b0
Compare
| List<ListOffsetsResultForBucket> listOffsetsResultForBuckets = future1.get(); | ||
| for (ListOffsetsResultForBucket result : listOffsetsResultForBuckets) { | ||
| if (result.getErrorCode() != 0) { | ||
| log.warn(result.getErrorMessage() + "," + result.getErrorCode()); |
There was a problem hiding this comment.
only used for debugging? can be removed now?
| RestartStrategyOptions.RESTART_STRATEGY, | ||
| RestartStrategyOptions.RestartStrategyType | ||
| .NO_RESTART_STRATEGY | ||
| .getMainValue())); |
There was a problem hiding this comment.
Why do we have to configure this?
|
@loserwang1024 I appended a commit to remove the unnecessary code. I will merge it if you have no concerns. |
Purpose
Linked issue: close #274