Hi Ammon, Unfortunately you're right. I think the Flink Kinesis Consumer specific configs, e.g. keys in the ConsumerConfigConstants class, were overlooked in the initial design.
One way to workaround this is to use the `SourceFunctionSpec` [1]. Using that spec, you can use any Flink SourceFunction (e.g. a FlinkKinesisConsumer) as the ingress. Simply instantiate a `SourceFunctionSpec` with the desired ID, and provide a custom FlinkKinesisConsumer that you create directly (which should allow you to provide the ConsumerConfigConstants). As a side note, I've created this JIRA to address the issue you encountered, as I believe this should be supported in the native StateFun Kinesis ingress [2]. Cheers, Gordon [1] https://github.com/apache/flink-statefun/blob/master/statefun-flink/statefun-flink-io/src/main/java/org/apache/flink/statefun/flink/io/datastream/SourceFunctionSpec.java [2] https://issues.apache.org/jira/browse/FLINK-22529 On Thu, Apr 29, 2021 at 7:25 AM Ammon Diether <adiet...@gmail.com> wrote: > > When using Flink Stateful Function's KinesisIngressBuilder, I do not see a > way to set things like ConsumerConfigConstants.SHARD_GETRECORDS_MAX or > ConsumerConfigConstants.SHARD_GETRECORDS_INTERVAL_MILLIS > > Looking at KinesisSourceProvider, it appears that this is the spot that > creates the FlinkKinesisConsumer. The function named > propertiesFromSpec(kinesisIngressSpec) only allows for AWS properties and a > few startup position properties. > ConsumerConfigConstants.SHARD_GETRECORDS_MAX cannot be provided. > > Is there an obvious workaround? > >
