Split Store and implementations from schema-registry #422
Conversation
|
Can one of the admins verify this patch? |
|
It looks like @Noddy76 hasn't signed our Contributor License Agreement, yet. Appreciation of efforts, clabot |
|
I have now signed the CLA at http://clabot.confluent.io/cla |
|
ok to test |
| </dependency> | ||
| <dependency> | ||
| <groupId>org.slf4j</groupId> | ||
| <artifactId>slf4j-log4j12</artifactId> |
There was a problem hiding this comment.
If this is purely a library rather than a service then normally you'd only use slf4j-api since you can't assume the library user will be using log4j.
| <!-- To prevent a UniqueResourceException due the same resource existing in both | ||
| org.apache.directory.api/api-all and org.apache.directory.api/api-ldap-schema-data, api-ldap-schema-data | ||
| needs to be excluded from all directory-related maven dependencies. --> | ||
| <dependency> |
There was a problem hiding this comment.
It'd be good to try to prune down dependencies in core/pom.xml since a bunch of stuff is included here as well. In particular, these bits stand out -- I think they are only needed for the SASL utility class that got moved, so probably the equivalent sections of core/pom.xml don't need all this duplicated Maven config.
| import org.apache.kafka.common.config.ConfigDef; | ||
| import org.apache.kafka.common.protocol.SecurityProtocol; | ||
|
|
||
| public class KafkaStoreConfig extends AbstractConfig { |
There was a problem hiding this comment.
Since it looks like SchemaRegistryConfig didn't change, I assume the obvious drawback here is lots of duplication. I'm guessing there wasn't a clean way to combine the configs since you now want to inherit from both RestConfig and KafkaStoreConfig in SchemaRegistryConfig? I'm just worried about these things falling out of sync, especially as it relates to documentation since we auto generate the configuration docs from the code.
| prop.put(SchemaRegistryConfig.KAFKASTORE_TOPIC_CONFIG, kafkaTopic); | ||
| prop.setProperty(KafkaStoreConfig.KAFKASTORE_CONFIG_PREFIX + KafkaStoreConfig.KAFKASTORE_CONNECTION_URL_CONFIG, zkConnect); | ||
| prop.put(KafkaStoreConfig.KAFKASTORE_CONFIG_PREFIX + KafkaStoreConfig.KAFKASTORE_TOPIC_CONFIG, kafkaTopic); | ||
| prop.setProperty(KafkaStoreConfig.KAFKASTORE_CONFIG_PREFIX + KafkaStoreConfig.KAFKASTORE_KAFKA_GROUPID, "test-group"); |
There was a problem hiding this comment.
This group ID seems to be replacing some setup we previously did in Kafka store, but this doesn't match the values it previously did for the schema registry service. This would be a compatibility issue since that ID is used to uniquely identify the instance. See comment later on for more details.
| config.getInt(SchemaRegistryConfig.KAFKASTORE_TOPIC_REPLICATION_FACTOR_CONFIG); | ||
| int port = KafkaSchemaRegistry.getPortForIdentity(config.getInt(SchemaRegistryConfig.PORT_CONFIG), | ||
| config.getList(RestConfig.LISTENERS_CONFIG)); | ||
| this.groupId = String.format("schema-registry-%s-%d", |
There was a problem hiding this comment.
This is where we seem to have lost some unique information about the schema registry instance. Since this is used as the consumer group ID, just using "test-group" in the other part of the patch isn't ideal. In practice, you can probably get away with this because we use the new consumer in "simple" consumer mode and don't commit offsets. However, this makes some logs much harder to use since they don't have unique identifying info. We probably want to maintain the group ID if possible.
| StoreUpdateHandler<K, V> storeUpdateHandler, | ||
| Serializer<K, V> serializer, | ||
| Store<K, V> localStore, | ||
| K noopKey) { | ||
| K noopKey, | ||
| boolean isZkSecurityEnabled) { |
There was a problem hiding this comment.
Why make this a constructor flag rather than a config given that everything else is passed in via config?
More generally, as this becomes truly public, supported API rather than internal implementation detail, we'll need to be careful about compatibility so we'd like to make sure these interfaces, especially the constructors and most used methods, are just what we want.
| protected List<KafkaServer> servers = null; | ||
| protected String brokerList = null; | ||
|
|
||
| //protected RestApp restApp = null; |
There was a problem hiding this comment.
Can we avoid a lot of code duplication here with a bit of inheritance and overrides? It seems like very little changed aside from removing a few lines related to instantiating and starting the rest app?
| @@ -0,0 +1,31 @@ | |||
| /* | |||
There was a problem hiding this comment.
It seems this could move from core rather than being copied. Doesn't look like it is used anymore since all the tests it was used in have moved.
| @@ -0,0 +1,73 @@ | |||
| /** | |||
There was a problem hiding this comment.
It seems this could move from core rather than being copied. Doesn't look like it is used anymore since all the tests it was used in have moved.
| this.serializer = serializer; | ||
| this.defaultCompatibilityLevel = config.compatibilityType(); | ||
| this.guidToSchemaKey = new HashMap<Integer, SchemaKey>(); | ||
| this.schemaHashToGuid = new HashMap<MD5, SchemaIdAndSubjects>(); | ||
| this.zkAclsEnabled = checkZkAclConfig(config); | ||
|
|
||
| KafkaStoreConfig kafkaStoreConfig = new KafkaStoreConfig(config.originalsWithPrefix(KafkaStoreConfig.KAFKASTORE_CONFIG_PREFIX)); |
There was a problem hiding this comment.
This isn't quite enough because you have at least the groupid that is now required for the KafkaStore but will not be included in the SchemaRegistryConfig. This ties back to the test-group issue -- this is where you would need to get the identity info that got dropped from KafkaStore.
Easy way to test this -- try to run schema registry with a simple default config :) Currently it will crash immediately because of this. (Generally, a good sanity check when making substantial changes like this is to just execute the entire quickstart.)
|
Hey @ewencp, thanks for the above feedback. Based on our e-mail discussion it sounds like there is similar code in the Kafka-connect code base that we could base this on. How do you recommend we proceed? Continue with fixing this according to your comments? Or take a look at the Kafka-connect code and modify this to rather use that approach? Also, if this "Kafka as a key-value store" functionality is being used from two Kafka related modules plus us having our own needs for something like this, wouldn't it make sense to make it a top-level module under the main Kafka project and change the other uses to use this common "library"? I think that would be the desired end state but appreciate we might want to take a phase approach to that. Thoughts? |
|
@massdosage Good question. Honestly, the best path will depend a lot on your needs. The fact that two modules have something related to your need (interactive queries (previously known as queryable state) in streams and My major concern with with suggesting focusing on those two libraries is that adding public API to Kafka requires KIPs and we've got multiple candidates with substantially different APIs, very different use cases, and I think pretty different requirements. That said, both the ones in Kafka are more modern and better implementations than KafkaStore here, which has to jump through some weird hoops to figure out that you have reached the end of the log (writing It's definitely easier to start with a (potentially unstable) API externally and maybe promote it eventually into a first class Kafka API. That said, I'm not sure what the reception to that will be given that folks are just starting to scratch the surface wrt using Kafka this way, i.e. I'm uncertain how comfortable people will be committing to supporting an interface when we have only a couple of examples of how it would be used. Take a look at how much discussion went into the new consumer API and behavior to get an idea for how involved this can be :) We've also been pretty conservative in adding APIs. For example, we've brought up the idea of a producer that logs locally to disk (for better durability in the event of connection failures combined with potential failures) because quite a few people want this, but this hasn't been added yet and we're not sure we want it as a core Kafka API. Out of curiosity, if you're using Kafka as a key-value store, do you have requirements on having a single writer per key? If so, how are you guaranteeing this? And do you have consistency requirements if you have multiple nodes serving requests? Do you need read-after-write semantics? There are a lot of details that need to be handled on top of what Kafka provides, so I'm curious about your exact requirements. |
|
@Noddy76 Based on the comments and the activity on the PR, my assumption is that this is no longer needed. With that assumption, I'm closing this PR. If you think otherwise, please feel free to reopen and we can work it through. |
This pull request splits
Store<K, V>and it's implementations from schema-registry into their own module for reuse elsewhere.Background
Since the implementation of
cleanup.policy = compactin Kafka you can use a topic as a distributed key-value store. Schema Registry does this to hold schema. When looking for a client library that implemented this functionality we found the implementation within schema-registry but it was coupled to the rest of the code and could not be used on it's own.