Hey fellow Flinkers I m currently working on an addition to Apache Flink #troubleshooting

Hey fellow Flinkers, I’m currently working on an ...

Rion Williams

06/22/2023, 1:34 AM

Hey fellow Flinkers, I’m currently working on an addition to an existing job that handles enriching events as they flow through from multiple streams. My need is to use some tenant-specific configuration to do so, which currently exists in a Kafka topic that’s populated via a CDC connector (and contains essentially the state/configuration for that tenant). In an attempt to avoid keying everything by tenant, I thought I could use Broadcast State to accomplish it through the following: • Create a new broadcast stream using a KafkaSource to read the CDC records and map them to a serializable object containing the enrichment that I need. • After performing a

union

of all of my other source streams (all mapped to a single common schema), connect that with the previous broadcast stream. • Apply the enrichments in an appropriate BroadcastProcessFunction where: ◦ Broadcast elements are simply stored in broadcast state (something like

context.getBroadcastState(descriptor).put(x.key, x.value)

); Not sure if there’s anything more needed. ◦ Non-Broadcast elements would simply check broadcast state for the given tenant configuration and apply the enrichment. Since I haven’t used Broadcast State previously, my questions are: • Would the addition of a new stream/join/broadcast like this be a breaking change? I.e. would the job be able to be restored from the previous state/savepoint prior to the change? • Is Broadcast State safe for a scenario like this? The cardinality of the data is fairly small, possibly a thousand or so, with the objects only containing two fields. Enough to store in memory pretty easily. • Does parallelism increase any issues here? My assumption is that for parallelism >1, performing the

broadcastState.put(…)

operation would rebroadcast that new/updated state to all of the task managers/tasks to ensure that the same data is eventually available. • Since Broadcast state is checkpointed, how does that factor in when the job is restarted or recovers from a failure? Will it be restored/available when data begins flowing again? I’m happy to provide any more context/detail/etc. if it helps!

Rion Williams

06/22/2023, 2:11 AM

Here’s a few code snippets that might help provide some context. This is the declaration of the tenant stream (uses a wrapper to construct the

KafkaSource

but it’s basically just taking the topic name and then subsequently mapping those values to objects:

Copy code

val tenantStream = streamEnv
            .fromSource(
                KafkaStringSource.build(parameterTool, parameterTool.get("tenant-topic")),
                WatermarkStrategy.noWatermarks(),
                "kafka-tenant-source"
            )
            .flatMap(MongoTenantMapper())
            .name("map-mongo-json-to-tenants")
            .uid("map-mongo-json-to-tenants")
            .broadcast(StateDescriptors.tenants)

The state descriptor itself is pretty simple and just has the tenant objects as a map:

Copy code

val tenants = MapStateDescriptor(
        "tenant-config-broadcast",
        String::class.java,
        FlinkTenant::class.java
)

And then when the previous streams are joined against the broadcasted tenant-stream:

Copy code

val commonEvents = stream.union(...,...,...)
            .connect(tenantStream)
            .process(TenantConfigurationEnricher())
            .name("tenant-configuration-enricher")
            .uid("tenant-configuration-enricher")

And finally the function itself (omitted for brevity):

Copy code

class TenantConfigurationEnricher : BroadcastProcessFunction<CommonEvent, FlinkElasticaTenant, CommonEvent>() {
    
    override fun processElement(event: CommonEvent, context: ReadOnlyContext, collector: Collector<CommonEvent>) {
        // Get the current broadcast state
        val tenantConfigurations = context.getBroadcastState(StateDescriptors.tenants)

        val tenant = event.tenantName
        if (tenantConfigurations.contains(tenant)) {
            val tenantConfiguration = tenantConfigurations.get(tenant)
            
            // Perform enrichment here
        }

        // Output the event (enriched or not)
        collector.collect(event)
    }

    override fun processBroadcastElement(tenant: FlinkElasticaTenant, context: Context, collector: Collector<CommonEvent>) {
        context.getBroadcastState(StateDescriptors.tenants).put(tenant.name, tenant)
    }
}

Rion Williams

06/22/2023, 2:12 AM

Does all of this seem reasonable? Or are there any major issues that jump out?

Open in Slack

Previous Next