initpos LATEST triggers InvalidArgumentException #13
Comments
cizmazia
changed the title
cizmazia
changed the title
cizmazia
changed the title
|
Hello, the 3 extra zeros do look suspicious indeed. It looks like if the extra 3 zeroes were not there, it would be ok. I have not seen this error before. Please confirm how you are running Flink, is it local/hosted or managed (AWS KDA etc)? To consume from If you are in a position to compile the connector from source, this is an option for now until we resolve the problem. You can modify the method here to simply At the moment I am struggling to reproduce the problem, so more information would be useful. |
awslabs/amazon-kinesis-connector-flink#13 Steps: docker-compose up -d export AWS_DEFAULT_REGION=eu-west-1 export AWS_ACCESS_KEY_ID=test export AWS_SECRET_ACCESS_KEY=test aws kinesis create-stream --endpoint-url=http://localhost:4566 --stream-name ExampleInputStream --shard-count 1 aws kinesis create-stream --endpoint-url=http://localhost:4566 --stream-name ExampleOutputStream --shard-count 1 mvn clean test
|
Thanks for getting back! Please take a look here on how to reproduce the issue: |
awslabs/amazon-kinesis-connector-flink#13 Steps: docker-compose up -d export AWS_DEFAULT_REGION=us-west-2 export AWS_ACCESS_KEY_ID=test export AWS_SECRET_ACCESS_KEY=test aws kinesis create-stream --endpoint-url=http://localhost:4566 --stream-name ExampleInputStream --shard-count 1 mvn clean test
awslabs/amazon-kinesis-connector-flink#13 Steps: docker-compose up -d export AWS_DEFAULT_REGION=us-west-2 export AWS_ACCESS_KEY_ID=test export AWS_SECRET_ACCESS_KEY=test aws kinesis create-stream --endpoint-url=http://localhost:4566 --stream-name ExampleInputStream --shard-count 1 mvn clean test
|
Thanks for the self contained reproduction, very nice. I am fairly certain it is a kineasalite issue, this looks like it will add three zeros: Since the AWS SDK uses a The issue is not observed when hitting the real Kinesis service. The kinesallite change was added in a commit indicating it is being fixed: It looks like Kinesalite was broken in v3.3.3 and Flink e2e tests use 3.3.1 so are unaffected: |
|
Thanks for tracking this down. Much appreciated! |
|
Related mhart/kinesalite#70 |
|
The breaking change has been introduced in kinesalite v1.12.0 |
|
Yeah you are right. The e2e test uses |
|
@dannycranmer I am running into a similar issue on the real Kinesis service when CBOR is disabled, but not when it is enabled (default). I've confirmed I get this error either when I set this property in my job: or when I set this environment variable, as they do in the Flink tests you linked above I know disabling CBOR is not required for the real Kinesis service, but it is for localstack, which uses kinesalite behind-the-scenes, which impacts the ability to run integration tests or for local development. I did a bit of digging, and I think the issue is related to how requests are serialized to JSON. What I did was setup a local proxy server to inspect messages being sent from my Flink job to Kinesalite. When calling Notice that the timestamp field is in milliseconds. The AWS API Reference says that this should be in seconds. So I think kinesalite multiplying by 1000 is in fact correct and the issue seems to be serialization of the timestamp when using the JSON API. |
|
Hey @davidtag. Thanks for the investigation and information. I will also take a look, since the Flink connector does not serialise the Record, I wonder if there is a bug in the AWS SDK. I will keep you posted. |
|
Thanks @dannycranmer. Any insight into where the issue may be? |
|
I conducted a simple test (SDK v1.11.844 and v1.11.1030) and it does look like an SDK or Kinesis issue. When I try to disable CBOR I see the following error, this is coming from the Kinesis service: Here is my test code, the case with CBOR enabled passes. I inspected the HTTP payload and in both cases the full timestamp (millis) is being submitted. I have raised an issue in the SDK project: aws/aws-sdk-java#2588 |
|
hi @dannycranmer |
|
Hey @SOD-DOB, the Apache Flink connector adds a layer of abstraction over other connectors to allow Flink to read/write from/to Kinesis Stream shards. In order to integrate any external data with Flink you must use a Flink source/sink interface to allow the systems to talk to each other. Typically Flink connector implementations are thin layers over off-the-shelf libraries, for example AWS SDK or Kafka SDK. For Kinesis, the following libraries are used:
The difference between this library and the Apache Flink Kinesis connector, is that we have backported some features for earlier versions of Apache Flink, in order to allow customers to leverage these features with Kinesis Data Analytics:
In terms of bug fix parity, we try our best to port bug fixes across all versions of Flink. You only need to use this connector if you want to use EFO or Table API, otherwise you can use the Apache connector. |
Hey, please could advise me on how to resolve the following issue?
With the configuration
flink.stream.initpos=LATEST, AWSUtil changes the configuration to AT_TIMESTAMP, starting from the current time. The Kinesis client is created byAWSUtil, however, it does not accept the timestamp and throwsInvalidArgumentException"The timestampInMillis parameter cannot be greater than the currentTimestampInMillis":1612801872966000(the 3 extra zeros look suspicious)1612801873011Versions
The text was updated successfully, but these errors were encountered: