AMashenkov commented on code in PR #1134:
URL: https://github.com/apache/ignite-3/pull/1134#discussion_r985823919
##########
modules/schema/src/main/java/org/apache/ignite/internal/schema/SchemaManager.java:
##########
@@ -84,51 +88,61 @@ public class SchemaManager extends Producer<SchemaEvent,
SchemaEventParameters>
/** Versioned store for tables by name. */
private final VersionedValue<Map<UUID, SchemaRegistryImpl>> registriesVv;
+ /** Meta storage manager. */
+ private final MetaStorageManager metastorageMgr;
+
/** Constructor. */
- public SchemaManager(Consumer<Function<Long, CompletableFuture<?>>>
registry, TablesConfiguration tablesCfg) {
+ public SchemaManager(
+ Consumer<Function<Long, CompletableFuture<?>>> registry,
+ TablesConfiguration tablesCfg,
+ MetaStorageManager metastorageMgr
+ ) {
this.registriesVv = new VersionedValue<>(registry, HashMap::new);
-
this.tablesCfg = tablesCfg;
+ this.metastorageMgr = metastorageMgr;
}
/** {@inheritDoc} */
@Override
public void start() {
- ((ExtendedTableConfiguration)
tablesCfg.tables().any()).schemas().listenElements(new
ConfigurationNamedListListener<>() {
- @Override
- public CompletableFuture<?>
onCreate(ConfigurationNotificationEvent<SchemaView> schemasCtx) {
- return onSchemaCreate(schemasCtx);
- }
- });
+ ((ExtendedTableConfiguration)
tablesCfg.tables().any()).schema().schema().listen(this::onSchemaCreate);
}
/**
* Listener of schema configuration changes.
*
- * @param schemasCtx Schemas configuration context.
+ * @param ctx Configuration context.
* @return A future.
*/
- private CompletableFuture<?>
onSchemaCreate(ConfigurationNotificationEvent<SchemaView> schemasCtx) {
+ private CompletableFuture<?>
onSchemaCreate(ConfigurationNotificationEvent<byte[]> ctx) {
if (!busyLock.enterBusy()) {
return failedFuture(new IgniteInternalException(NODE_STOPPING_ERR,
new NodeStoppingException()));
}
try {
- long causalityToken = schemasCtx.storageRevision();
+ final byte[] serialized =
ctx.config(ExtendedTableConfiguration.class).schema().schema().value();
- ExtendedTableConfiguration tblCfg =
schemasCtx.config(ExtendedTableConfiguration.class);
+ ExtendedTableConfiguration tblCfg =
ctx.config(ExtendedTableConfiguration.class);
+
+ long causalityToken = ctx.storageRevision();
UUID tblId = tblCfg.id().value();
String tableName = tblCfg.name().value();
- SchemaDescriptor schemaDescriptor =
SchemaSerializerImpl.INSTANCE.deserialize((schemasCtx.newValue().schema()));
+ SchemaDescriptor schemaDescriptor =
SchemaSerializerImpl.INSTANCE.deserialize(serialized);
Review Comment:
Let's catch serialisation (and possible other) exceptions and pass them to
resulting future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
