decodableco/decodable-pipeline-sdk

An SDK for implementing Flink jobs based on Decodable

decodable-pipeline-sdk's Issues

Automatically release staging repo during release process

Set up deployment to Maven Central

Misc. README improvements

Add Maven wrapper

See https://maven.apache.org/wrapper/

Simplify environment configuration in tests

Implement basic Decodable SDK

Providing testing utilities to pre-seed input streams and assert expected records on output streams

Update project version in example during release

Set up GitHub Actions for CI

Rename artifact to "decodable-pipeline-sdk"

Create an interface that enforces predictable operator names for the Decodable stream sources and sinks

• Currently, the Decodable stream sources and sinks, when inserted into a DataStream graph, are arbitrarily nameable. This presents issues for the Decodable platform when attempting to correlate runtime metrics from the graph operators back to the referenced streams.

Integrate with Flink's schema system for using other formats than String

Table API Support

It should be possible to refer to Decodable Streams via Flink SQL, using the CREATE TABLE ... syntax, as used for any other connectors.

Only execute tests if docker environment is present

Users w/o docker will see failing tests.

Add AWS IAM library as a dependency

It is required when running a job against Decodable.

Metrics support

Support key (de-)serializers

See #1 (comment)

Provide Maven archetype for bootstrapping custom pipeline projects

Upgrade to Flink 1.16.2

Secrets support

Log a warning when using undeclared sources and sinks

DecodableStreamSource and DecodableStreamSink should raise a warning in the logs when a stream is referenced which isn't declared in META-INF/decodable/stream-names.properties.

Expand README

Set up Maven enforcer plug-in to make sure Java 11 is used for the example

ClassCastException: class org.apache.flink.connector.kafka.sink.KafkaCommittable cannot be cast to class co.decodable.sdk.pipeline.DecodableCommittable

Via @rmetzger:

java.lang.Exception: Could not perform checkpoint 35 for operator Source: Purchase Orders Source -> Map -> Sink: Writer -> Sink: Committer (1/1)#8.
	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointAsyncInMailbox(StreamTask.java:1166)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$triggerCheckpointAsync$12(StreamTask.java:1113)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
	at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:398)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:367)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:352)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:831)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:780)
	at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:935)
	at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:914)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:728)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:550)
	at java.base/java.lang.Thread.run(Unknown Source)
Caused by: org.apache.flink.runtime.checkpoint.CheckpointException: Could not complete snapshot 35 for operator Source: Purchase Orders Source -> Map -> Sink: Writer -> Sink: Committer (1/1)#8. Failure reason: Checkpoint was declined.
	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:269)
	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:173)
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:345)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.checkpointStreamOperator(RegularOperatorChain.java:228)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.buildOperatorSnapshotFutures(RegularOperatorChain.java:213)
	at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.snapshotState(RegularOperatorChain.java:192)
	at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.takeSnapshotSync(SubtaskCheckpointCoordinatorImpl.java:726)
	at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:363)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$13(StreamTask.java:1281)
	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:1269)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointAsyncInMailbox(StreamTask.java:1154)
	... 14 more
Caused by: java.lang.ClassCastException: class org.apache.flink.connector.kafka.sink.KafkaCommittable cannot be cast to class co.decodable.sdk.pipeline.DecodableCommittable (org.apache.flink.connector.kafka.sink.KafkaCommittable and co.decodable.sdk.pipeline.DecodableCommittable are in unnamed module of loader org.apache.flink.util.ChildFirstClassLoader @563cd0e6)
	at co.decodable.sdk.pipeline.internal.DelegatingCommittableSerializer.serialize(DelegatingCommittableSerializer.java:14)
	at org.apache.flink.core.io.SimpleVersionedSerialization.writeVersionAndSerialize(SimpleVersionedSerialization.java:59)
	at org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer$SubtaskSimpleVersionedSerializer$RequestSimpleVersionedSerializer.serialize(CommittableCollectorSerializer.java:252)
	at org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer$SubtaskSimpleVersionedSerializer$RequestSimpleVersionedSerializer.serialize(CommittableCollectorSerializer.java:242)
	at org.apache.flink.core.io.SimpleVersionedSerialization.writeVersionAndSerializeList(SimpleVersionedSerialization.java:91)
	at org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer$SubtaskSimpleVersionedSerializer.serialize(CommittableCollectorSerializer.java:214)
	at org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer$SubtaskSimpleVersionedSerializer.serialize(CommittableCollectorSerializer.java:182)
	at org.apache.flink.core.io.SimpleVersionedSerialization.writeVersionAndSerializeList(SimpleVersionedSerialization.java:91)
	at org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer$CheckpointSimpleVersionedSerializer.serialize(CommittableCollectorSerializer.java:138)
	at org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer$CheckpointSimpleVersionedSerializer.serialize(CommittableCollectorSerializer.java:126)
	at org.apache.flink.core.io.SimpleVersionedSerialization.writeVersionAndSerializeList(SimpleVersionedSerialization.java:91)
	at org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer.serializeV2(CommittableCollectorSerializer.java:99)
	at org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer.serialize(CommittableCollectorSerializer.java:71)
	at org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer.serialize(CommittableCollectorSerializer.java:43)
	at org.apache.flink.core.io.SimpleVersionedSerialization.writeVersionAndSerialize(SimpleVersionedSerialization.java:174)
	at org.apache.flink.streaming.api.operators.util.SimpleVersionedListState.serialize(SimpleVersionedListState.java:102)
	at org.apache.flink.streaming.api.operators.util.SimpleVersionedListState.serializeAll(SimpleVersionedListState.java:113)
	at org.apache.flink.streaming.api.operators.util.SimpleVersionedListState.update(SimpleVersionedListState.java:71)
	at org.apache.flink.streaming.runtime.operators.sink.CommitterOperator.snapshotState(CommitterOperator.java:129)
	at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.snapshotState(StreamOperatorStateHandler.java:222)
	... 25 more

Provide means to expose stream names accessed by a custom pipeline

Publish API docs