Enable Filters to access shared state per virtual cluster.

[robobario]

Currently users can store state in their Filter, or perhaps use ThreadLocals or static state to share access to some threadsafe objects. If we wanted to scope that state to be shared only within one VirtualCluster then we currently don't have a mechanism to know what virtual cluster the current context is happening in.

A concrete example is topic encryption, we may have a set of topic policies and key manager definitions that are relevant to one Virtual Cluster only (because the targeted topic names could be used in another unrelated virtual cluster). We could potentially share an Encryption Module and a map from topic id to topic name across instances of the encrypt/decrypt filters and save having to populate them all the time.

Some ideas.

1. Make the virtual cluster name available in the filter context. Something like context.metadata().virtualClusterName() or context.virtualCluster().name(). Then the Filter would implement it's own static ConcurrentHashMap<String, State>
2. Enable state to be associated with the virtual cluster through the context, like context.state().virtualCluster().getOrCreate("xyz", () -> { new State()}, State.class)

Relevant to kroxylicious/topic-encryption#2

[robobario]

Or we could make Filters naturally scoped to a VirtualCluster so it's not something they pick up from the Context while filtering but something known at FIlter construction time?

[SamBarker]

I had been envisaging doing something like 1 as a bootstrap exercise when I self assigned kroxylicious/topic-encryption#2 to see how that played out in practice. Probably doing something along the same lines for a another filter or two to try and establish a common pattern before looking at something like 2.

[robobario]

Sweet (to be clear I'm not working on this, just poking the conversation)

[SamBarker]

Having said that something like
context.state().virtualCluster().getOrCreate("xyz", () -> { new State()}, State.class) sounds like a sane starting point but I wonder if we might think about

enum Scope { CONNECTION, TOPIC, VIRTUAL_CLUSTER, TARGET_CLUSTER };
context.state(scope).getOrCreate("xyz", () -> { new State()}, State.class))

JUnit has a similar state model we can look to for inspiration as well.

[k-wall]

I like the look of that suggestion @SamBarker. It got me thinking about Netty's type safe attributes as a source of inspiration too.

    protected static final AttributeKey<Map<RoutingKey, VirtualClusterBinding>> CHANNEL_BINDINGS = AttributeKey.newInstance("channelBindings");

[k-wall]

I'm not sure about topic scoped state. What would be the use-case? The other scopes make sense to me.

[franz1981]

FYI, for Vert-x I've modified an existing write-combining primitive construct to be more "thread-per-core" non-blocking friendly, see
https://github.com/eclipse-vertx/vert.x/blob/7bc52114b531d5afe68110e8eff08ccc97542a6b/src/main/java/io/vertx/core/net/impl/pool/CombinerExecutor.java#L26

TLDR it's a sort of cooperative lock where the winner to enter the lock execute concurrent/racy actions submitted by other threads.
This is useful in heavyly modified consistent states use cases.
The vertx case requires running post-actions after the main actions, out of the mutual exclusion regions, but the actual pattern is pretty simple, without such post-actions (simpler than a lock, actually).
Other frameworks instead decide to elect a "event loop" owner for the shared state, which allow (wait)lock-free read access (using versions to be sure data is not invalid/stale) and a thread/event-loop confined mechanism to submit (rare) modification to the shared state. These are viable patterns in case of:

• combiner: heavy write of shared state scenario
• confined (require using the event loop executor if modified by other event loops): rare writes of shared state scenario

in both cases, read path can be wait free

[SamBarker]

I hadn't even got to thinking about dynamically sharing state and the threading implications but your executor certainly sounds like something we should consider when implementing it the store.

[tombentley]

Mutable shared state is hard. The use case of wanting to cache information per topic is likely to be common. There might be similar cases for storing information about things like consumer groups.

Here are some random thoughts which may, or may not be useful:

• I don't think anyone's said it explicitly, but I assume we're all thinking that this shared state is local to a filter implementation class. I.e. we don't want to let FooFilter observe (and potentially overwrite) state written by BarFilter, but we do want to allow two instances of FooFilter (in the same virtual cluster) to share state. And actually even that is logically a bit suspect: It's certainly an edge case, but if we had the same filter implementation type at two points on the chain should they be able to communicate via this shared state? I suspect not.
• I think we should be opinionated about the relative frequency of writes vs reads, and my hunch is that we should prioritize reads, because I think that's likely to be a more common case, than write heavy. I like the direction that Franz is going with lock/wait free get().
• I can see the attraction of "scopes", but they each imply a different lifetime and questions arise about how we cope with the lifecycle. Some scopes are simpler than others from an implementation PoV:
  • Removing a virtual or target cluster looks like a relatively uncommon thing to do and it's local to the proxy, so they're probably relatively easy (threading issues aside).
  • Topic scope is tricky,
    • if there is non-proxied access to the target cluster then you won't observe DeleteTopicResponse. Ditto if there are multiple instances of the proxy (only one would observe DeleteTopicResponse). Both of these imply the possibility of stale state remaining in a per-topic cache.
    • even in the case where we can guarantee observing DeleteTopicResponse in order to remove state associated with a topic I think there is a race:
      1. Filter instance A observes DeleteTopicResponse from broker 1 for topic with id x
      2. Filter instance B observes FetchResponse for topic with id x
      3. Filter instance B doesn't find that id, so makes a MetadataRequest to broker 2.
      4. Broker 2 serves up the metadata for topic with id x
      5. Broker 2 updates it metadata cache from controller 3.
      6. Filter instance B updates the cache with information about topic with id x.
  • While topic churn in the cluster is unlikely to lead to OOME in a proxy process (because topics are usually relatively long-lived things) a similar thing might happen with consumer groups where uses often omit specifying a group.id for ephemeral consumers.
  • So we need to define the contract shared topic (and other state) in a way such that the state is not required for correctness. In the topic encryption use case, we could safely expire the topic id from the cache (with or without an explicit revalidation step).
  • Consequently perhaps we don't want to model scope as an enum (where the different contracts of the state sharing are not going to be very obvious), and prefer something like context.topicCache().getOrCreate(...).
  • Similarly, I would like better motivation for the virtual cluster scope and target cluster scopes needing mutable shared state (as opposed to static, coming from the configuration). If we can't see any valid use cases right now we might want to expose only a get(), rather than a getOrCreate().

[franvila]

We don't have time to address this, so let's close it.

[gracegrimwood]

Possible duplicate of #1238