Flink Connector not compatible with Flink 1.15.2

[nicusX]

The Flink Connector does not seem to be compatible with Flink 1.15.2 (the Flink version now supported by Amazon Kinesis Data Analytics).

When initialising the sink it throws the following exception:

java.lang.NoSuchMethodError: 'org.apache.flink.metrics.MetricGroup org.apache.flink.api.connector.sink.Sink$InitContext.metricGroup()'
	at com.amazonaws.samples.connectors.timestream.TimestreamSinkWriter.createSinkMetricGroup(TimestreamSinkWriter.java:77)
	at imported.vnext.org.apache.flink.connector.base.sink.sink.writer.AsyncSinkWriter.<init>(AsyncSinkWriter.java:248)
	at com.amazonaws.samples.connectors.timestream.TimestreamSinkWriter.<init>(TimestreamSinkWriter.java:56)
	at com.amazonaws.samples.connectors.timestream.TimestreamSink.createWriter(TimestreamSink.java:51)
	at org.apache.flink.streaming.api.transformations.SinkV1Adapter.createWriter(SinkV1Adapter.java:77)
	at org.apache.flink.streaming.api.transformations.SinkV1Adapter.createWriter(SinkV1Adapter.java:56)
	at org.apache.flink.streaming.runtime.operators.sink.StatelessSinkWriterStateHandler.createWriter(StatelessSinkWriterStateHandler.java:39)
	at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.initializeState(SinkWriterOperator.java:146)
	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:122)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:286)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:106)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:700)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:676)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:643)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:948)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:917)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:741)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:563)
	at java.base/java.lang.Thread.run(Thread.java:832)

[cmoetzing]

Had a similar issue. Wasn't sure which version the main branch would support so I upgraded to a more recent version on my own branch. Maybe this helps.

[nicusX]

Thanks @cmoetzing. I did the same locally. But I guess a real upgrade would require moving away from the deprecated Sink API to the new one

[alec-ferguson-bppulse]

Seeing the same issue. It looks like the latest mainline branch references sink2. Curious if the maintainers plan to publish a new version soon? Looking forward to using it, looks like some good changes in there.

[PeterCsalaHbo]

Related PR: https://github.com/awslabs/amazon-timestream-tools/pull/124

[vzateychuk]

Dear colleagues, @dannycranmer, @PeterCsalaHbo, I'm still having the same issue (NoSuchMethodError: 'MetricGroup Sink$InitContext.metricGroup()') on AWS Kinesis with Flink application ver:1.15.3 even related PR:#124 already merged. Seems to me that the deprecated Sink$InitContext is still in use in flink-connector-timestream ver:0.2. I wonder if you could give any piece of advice how to tackle the problem? Is upgrade flink to 1.16 can handle the problem?

[dannycranmer]

Hey @vzateychuk, I believe you should be using v0.3, did you try that?

[vzateychuk]

Hey @vzateychuk, I believe you should be using v0.3, did you try that?

@dannycranmer thank you for a hint, I've been waiting v0.3 for a week or two, hope it will solve the problem.

[dkovalenkoI]

@nicusX , The pull request has been created to address the reported issue.

Thank you for your patience and valuable feedback.

[dannycranmer]

I am a little confused by the "fix" here. The issue reported is "Flink Connector not compatible with Flink 1.15.2" and we have "fixed" it by upgrading from Flink 1.15 to 1.17?

        <flink.version>1.15.3</flink.version>
        <flink.version>1.17.1</flink.version>

Did you verify this update is compatible with Flink 1.15? I can confirm the v0.3 is compatible with Flink 1.15.