import io.airlift.json.JsonModule; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> config, ConnectorContext context)
{
    requireNonNull(config, "config is null");

    try {
        Bootstrap app = new Bootstrap(
                new JsonModule(),
                new HDFSModule(connectorId, context.getTypeManager())
        );

        Injector injector = app
                .strictConfig()
                .doNotInitializeLogging()
                .setRequiredConfigurationProperties(config)
                .initialize();

        return injector.getInstance(HDFSConnector.class);
    }
    catch (Exception e) {
        e.printStackTrace();
    }
    return null;
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@Override
public Connector create(final String connectorId, Map<String, String> requiredConfig)
{
    requireNonNull(requiredConfig, "requiredConfig is null");
    requireNonNull(optionalConfig, "optionalConfig is null");

    try {
        // A plugin is not required to use Guice; it is just very convenient
        Bootstrap app = new Bootstrap(
                new JsonModule(),
                new ExampleModule(connectorId, typeManager));

    Injector injector = app
                .strictConfig()
                .doNotInitializeLogging()
                .setRequiredConfigurationProperties(requiredConfig)
                .setOptionalConfigurationProperties(optionalConfig)
                .initialize();

        return injector.getInstance(ExampleConnector.class);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@Override
public Connector create(final String connectorId, Map<String, String> requiredConfig) {
    checkNotNull(requiredConfig, "requiredConfig is null");
    checkNotNull(optionalConfig, "optionalConfig is null");

    try {
        // A plugin is not required to use Guice; it is just very convenient
        Bootstrap app = new Bootstrap(
                new JsonModule(),
                new RiakModule(connectorId, typeManager));

        Injector injector = app
                .strictConfig()
                .doNotInitializeLogging()
                .setRequiredConfigurationProperties(requiredConfig)
                .setOptionalConfigurationProperties(optionalConfig)
                .initialize();

        return injector.getInstance(RiakConnector.class);
    } catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
public static void main(String[] args)
        throws Exception
{
    if (System.getProperty("config", null) == null) {
        System.setProperty("config", "etc/agent.properties");
    }

    Bootstrap bootstrap = new Bootstrap(
            new NodeModule(),
            new DiscoveryModule(),
            new HttpServerModule(),
            new JsonModule(),
            new JaxrsModule(true), // requireExplicitBindings = true
            new HttpEventModule(),
            new JsonReaderModule(false),
            new DynamicAnnouncementModule(),
            new AgentServerModule()
    );

    try {
        Injector injector = bootstrap.strictConfig().initialize();
        injector.getInstance(Announcer.class).start();
    }
    catch (Exception e) {
        LOG.error(e, "Error starting server");
    }
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
public static void main(String[] args)
        throws Exception
{
    if (System.getProperty("config", null) == null) {
        System.setProperty("config", "etc/controller.properties");
    }

    Bootstrap bootstrap = new Bootstrap(
            new NodeModule(),
            new DiscoveryModule(),
            new DiscoveryServerModule(),
            new HttpEventModule(),
            new HttpServerModule(),
            new JsonModule(),
            new JaxrsModule(true), // requireExplicitBindings = true'
            new JmxModule(),
            new MBeanModule(),
            new JsonReaderModule(false),
            new ControllerServerModule()
    );

    try {
        bootstrap.strictConfig().initialize();
    }
    catch (Exception e) {
        LOG.error(e, "Error starting server");
    }
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> config)
{
    requireNonNull(connectorId, "connectorId is null");
    requireNonNull(config, "config is null");

    try {
        Bootstrap app = new Bootstrap(
                new JsonModule(),
                new KafkaConnectorModule(),
                binder -> {
                    binder.bind(KafkaConnectorId.class).toInstance(new KafkaConnectorId(connectorId));
                    binder.bind(TypeManager.class).toInstance(typeManager);
                    binder.bind(NodeManager.class).toInstance(nodeManager);

                    if (tableDescriptionSupplier.isPresent()) {
                        binder.bind(new TypeLiteral<Supplier<Map<SchemaTableName, KafkaTopicDescription>>>() {}).toInstance(tableDescriptionSupplier.get());
                    }
                    else {
                        binder.bind(new TypeLiteral<Supplier<Map<SchemaTableName, KafkaTopicDescription>>>() {}).to(KafkaTableDescriptionSupplier.class).in(Scopes.SINGLETON);
                    }
                }
        );

        Injector injector = app.strictConfig()
                .doNotInitializeLogging()
                .setRequiredConfigurationProperties(config)
                .setOptionalConfigurationProperties(optionalConfig)
                .initialize();

        return injector.getInstance(KafkaConnector.class);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> config)
{
    try {
        Bootstrap app = new Bootstrap(
                new JsonModule(),
                new MBeanModule(),
                binder -> {
                    CurrentNodeId currentNodeId = new CurrentNodeId(nodeManager.getCurrentNode().getNodeIdentifier());
                    MBeanServer mbeanServer = new RebindSafeMBeanServer(getPlatformMBeanServer());

                    binder.bind(MBeanServer.class).toInstance(mbeanServer);
                    binder.bind(CurrentNodeId.class).toInstance(currentNodeId);
                    binder.bind(NodeManager.class).toInstance(nodeManager);
                    binder.bind(PageSorter.class).toInstance(pageSorter);
                    binder.bind(BlockEncodingSerde.class).toInstance(blockEncodingSerde);
                    binder.bind(TypeManager.class).toInstance(typeManager);
                },
                metadataModule,
                new BackupModule(backupProviders),
                new StorageModule(connectorId),
                new RaptorModule(connectorId));

        Injector injector = app
                .strictConfig()
                .doNotInitializeLogging()
                .setRequiredConfigurationProperties(config)
                .setOptionalConfigurationProperties(optionalConfig)
                .initialize();

        return injector.getInstance(RaptorConnector.class);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> config)
{
    requireNonNull(connectorId, "connectorId is null");
    requireNonNull(config, "config is null");

    try {
        Bootstrap app = new Bootstrap(
                new JsonModule(),
                new RedisConnectorModule(),
                binder -> {
                    binder.bind(RedisConnectorId.class).toInstance(new RedisConnectorId(connectorId));
                    binder.bind(TypeManager.class).toInstance(typeManager);
                    binder.bind(NodeManager.class).toInstance(nodeManager);

                    if (tableDescriptionSupplier.isPresent()) {
                        binder.bind(new TypeLiteral<Supplier<Map<SchemaTableName, RedisTableDescription>>>() {}).toInstance(tableDescriptionSupplier.get());
                    }
                    else {
                        binder.bind(new TypeLiteral<Supplier<Map<SchemaTableName, RedisTableDescription>>>() {})
                                .to(RedisTableDescriptionSupplier.class)
                                .in(Scopes.SINGLETON);
                    }
                }
        );

        Injector injector = app.strictConfig()
                .doNotInitializeLogging()
                .setRequiredConfigurationProperties(config)
                .setOptionalConfigurationProperties(optionalConfig)
                .initialize();

        return injector.getInstance(RedisConnector.class);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
public TestingDiscoveryServer(String environment)
        throws Exception
{
    tempDir = Files.createTempDir();

    Map<String, String> serverProperties = ImmutableMap.<String, String>builder()
            .put("static.db.location", tempDir.getAbsolutePath())
            .put("discovery.store-cache-ttl", "0ms")
            .build();

    Bootstrap app = new Bootstrap(
            new MBeanModule(),
            new TestingNodeModule(environment),
            new TestingHttpServerModule(),
            new JsonModule(),
            new JaxrsModule(true),
            new DiscoveryServerModule(),
            new DiscoveryModule(),
            new TestingJmxModule());

    Injector injector = app
            .strictConfig()
            .doNotInitializeLogging()
            .setRequiredConfigurationProperties(serverProperties)
            .initialize();

    lifeCycleManager = injector.getInstance(LifeCycleManager.class);

    server = injector.getInstance(TestingHttpServer.class);
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@BeforeMethod
public void startUp()
{
    Injector injector = Guice.createInjector(new JsonModule(), new HandleJsonModule());

    objectMapper = injector.getInstance(ObjectMapper.class);
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> config)
{
    requireNonNull(config, "config is null");

    try {
        Bootstrap app = new Bootstrap(
                new MBeanModule(),
                new JsonModule(),
                new CassandraClientModule(connectorId),
                new Module()
                {
                    @Override
                    public void configure(Binder binder)
                    {
                        MBeanServer platformMBeanServer = ManagementFactory.getPlatformMBeanServer();
                        binder.bind(MBeanServer.class).toInstance(new RebindSafeMBeanServer(platformMBeanServer));
                    }
                });

        Injector injector = app.strictConfig().doNotInitializeLogging()
                .setRequiredConfigurationProperties(config)
                .setOptionalConfigurationProperties(optionalConfig).initialize();

        return injector.getInstance(CassandraConnector.class);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@Override
public void run()
{
    verifyJvmRequirements();
    verifySystemTimeIsReasonable();

    Logger log = Logger.get(PrestoServer.class);

    ImmutableList.Builder<Module> modules = ImmutableList.builder();
    modules.add(
            new NodeModule(),
            new DiscoveryModule(),
            new HttpServerModule(),
            new JsonModule(),
            new JaxrsModule(true),
            new MBeanModule(),
            new JmxModule(),
            new JmxHttpModule(),
            new LogJmxModule(),
            new TraceTokenModule(),
            new JsonEventModule(),
            new HttpEventModule(),
            new EmbeddedDiscoveryModule(),
            new ServerSecurityModule(),
            new AccessControlModule(),
            new ServerMainModule(sqlParserOptions),
            new GracefulShutdownModule(),
            installModuleIf(
                    NodeSchedulerConfig.class,
                    config -> LEGACY_NETWORK_TOPOLOGY.equalsIgnoreCase(config.getNetworkTopology()),
                    binder -> binder.bind(NetworkTopology.class).to(LegacyNetworkTopology.class).in(Scopes.SINGLETON)),
            installModuleIf(
                    NodeSchedulerConfig.class,
                    config -> "flat".equalsIgnoreCase(config.getNetworkTopology()),
                    binder -> binder.bind(NetworkTopology.class).to(FlatNetworkTopology.class).in(Scopes.SINGLETON))

    );

    modules.addAll(getAdditionalModules());

    Bootstrap app = new Bootstrap(modules.build());

    try {
        Injector injector = app.strictConfig().initialize();

        injector.getInstance(PluginManager.class).loadPlugins();

        injector.getInstance(CatalogManager.class).loadCatalogs();

        // TODO: remove this huge hack
        updateDatasources(
                injector.getInstance(Announcer.class),
                injector.getInstance(Metadata.class),
                injector.getInstance(ServerConfig.class),
                injector.getInstance(NodeSchedulerConfig.class));

        injector.getInstance(AccessControlManager.class).loadSystemAccessControl();

        injector.getInstance(Announcer.class).start();

        log.info("======== SERVER STARTED ========");
    }
    catch (Throwable e) {
        log.error(e);
        System.exit(1);
    }
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@Test
public void testExcludesCurrentNode()
        throws Exception
{
    Bootstrap app = new Bootstrap(
            new TestingNodeModule(),
            new TestingJmxModule(),
            new TestingDiscoveryModule(),
            new TestingHttpServerModule(),
            new TraceTokenModule(),
            new JsonModule(),
            new JaxrsModule(true),
            new FailureDetectorModule(),
            new Module()
            {
                @Override
                public void configure(Binder binder)
                {
                    configBinder(binder).bindConfig(QueryManagerConfig.class);
                    discoveryBinder(binder).bindSelector("presto");
                    discoveryBinder(binder).bindHttpAnnouncement("presto");

                    // Jersey with jetty 9 requires at least one resource
                    // todo add a dummy resource to airlift jaxrs in this case
                    jaxrsBinder(binder).bind(FooResource.class);
                }
            });

    Injector injector = app
            .strictConfig()
            .doNotInitializeLogging()
            .initialize();

    ServiceSelector selector = injector.getInstance(Key.get(ServiceSelector.class, serviceType("presto")));
    assertEquals(selector.selectAllServices().size(), 1);

    HeartbeatFailureDetector detector = injector.getInstance(HeartbeatFailureDetector.class);
    detector.updateMonitoredServices();

    assertEquals(detector.getTotalCount(), 0);
    assertEquals(detector.getActiveCount(), 0);
    assertEquals(detector.getFailedCount(), 0);
    assertTrue(detector.getFailed().isEmpty());
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> config)
{
    requireNonNull(config, "config is null");

    try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
        Bootstrap app = new Bootstrap(
                new NodeModule(),
                new MBeanModule(),
                new JsonModule(),
                new HiveClientModule(connectorId, metastore, typeManager, pageIndexerFactory),
                installModuleIf(
                        SecurityConfig.class,
                        security -> ALLOW_ALL_ACCESS_CONTROL.equalsIgnoreCase(security.getSecuritySystem()),
                        new NoSecurityModule()),
                installModuleIf(
                        SecurityConfig.class,
                        security -> "read-only".equalsIgnoreCase(security.getSecuritySystem()),
                        new ReadOnlySecurityModule()),
                installModuleIf(
                        SecurityConfig.class,
                        security -> "sql-standard".equalsIgnoreCase(security.getSecuritySystem()),
                        new SqlStandardSecurityModule()),
                binder -> {
                    MBeanServer platformMBeanServer = ManagementFactory.getPlatformMBeanServer();
                    binder.bind(MBeanServer.class).toInstance(new RebindSafeMBeanServer(platformMBeanServer));
                }
        );

        Injector injector = app
                .strictConfig()
                .doNotInitializeLogging()
                .setRequiredConfigurationProperties(config)
                .setOptionalConfigurationProperties(optionalConfig)
                .initialize();

        LifeCycleManager lifeCycleManager = injector.getInstance(LifeCycleManager.class);
        ConnectorMetadata metadata = injector.getInstance(ConnectorMetadata.class);
        ConnectorSplitManager splitManager = injector.getInstance(ConnectorSplitManager.class);
        ConnectorPageSourceProvider connectorPageSource = injector.getInstance(ConnectorPageSourceProvider.class);
        ConnectorPageSinkProvider pageSinkProvider = injector.getInstance(ConnectorPageSinkProvider.class);
        HiveSessionProperties hiveSessionProperties = injector.getInstance(HiveSessionProperties.class);
        HiveTableProperties hiveTableProperties = injector.getInstance(HiveTableProperties.class);
        ConnectorAccessControl accessControl = injector.getInstance(ConnectorAccessControl.class);

        return new HiveConnector(
                lifeCycleManager,
                new ClassLoaderSafeConnectorMetadata(metadata, classLoader),
                new ClassLoaderSafeConnectorSplitManager(splitManager, classLoader),
                new ClassLoaderSafeConnectorPageSourceProvider(connectorPageSource, classLoader),
                new ClassLoaderSafeConnectorPageSinkProvider(pageSinkProvider, classLoader),
                ImmutableSet.of(),
                hiveSessionProperties.getSessionProperties(),
                hiveTableProperties.getTableProperties(),
                accessControl);
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@Override
public Connector create(String connectorId, Map<String, String> config, ConnectorContext context)
{
    log.info("In connector factory create method.  Connector id: " + connectorId);
    requireNonNull(connectorId, "connectorId is null");
    requireNonNull(config, "config is null");

    try {
        Bootstrap app = new Bootstrap(
                new JsonModule(),
                new KinesisConnectorModule(),
                binder -> {
                    binder.bindConstant().annotatedWith(Names.named("connectorId")).to(connectorId);
                    binder.bind(ConnectorId.class).toInstance(new ConnectorId(connectorId));
                    binder.bind(TypeManager.class).toInstance(context.getTypeManager());
                    binder.bind(NodeManager.class).toInstance(context.getNodeManager());
                    // Note: moved creation from KinesisConnectorModule because connector manager accesses it earlier!
                    binder.bind(KinesisHandleResolver.class).toInstance(new KinesisHandleResolver(connectorName));

                    // Moved creation here from KinesisConnectorModule to make it easier to parameterize
                    if (altProviderClass.isPresent()) {
                        binder.bind(KinesisClientProvider.class).to(altProviderClass.get()).in(Scopes.SINGLETON);
                    }
                    else {
                        binder.bind(KinesisClientProvider.class).to(KinesisClientManager.class).in(Scopes.SINGLETON);
                    }

                    if (tableDescriptionSupplier.isPresent()) {
                        binder.bind(new TypeLiteral<Supplier<Map<SchemaTableName, KinesisStreamDescription>>>() {}).toInstance(tableDescriptionSupplier.get());
                    }
                    else {
                        binder.bind(new TypeLiteral<Supplier<Map<SchemaTableName, KinesisStreamDescription>>>() {}).to(KinesisTableDescriptionSupplier.class).in(Scopes.SINGLETON);
                    }
                }
        );

        this.injector = app.strictConfig()
                    .doNotInitializeLogging()
                    .setRequiredConfigurationProperties(config)
                    .setOptionalConfigurationProperties(optionalConfig)
                    .initialize();

        KinesisConnector connector = this.injector.getInstance(KinesisConnector.class);

        // Register objects for shutdown, at the moment only KinesisTableDescriptionSupplier
        if (!tableDescriptionSupplier.isPresent()) {
            // This will shutdown related dependent objects as well:
            KinesisTableDescriptionSupplier supp = getTableDescSupplier(this.injector);
            connector.registerShutdownObject(supp);
        }

        log.info("Done with injector.  Returning the connector itself.");
        return connector;
    }
    catch (Exception e) {
        throw Throwables.propagate(e);
    }
}
 

import io.airlift.json.JsonModule; //导入依赖的package包/类
@Override
    public Connector create(String connectorId, Map<String, String> config)
    {
        checkNotNull(config, "config is null");

        try {
            KafkaClientModule kafkaClientModule = new KafkaClientModule(connectorId);

            Bootstrap app = new Bootstrap(
                    new NodeModule(),
                    new JsonModule(),
                    kafkaClientModule
                );

            Injector injector = app.strictConfig().doNotInitializeLogging()
                    .setRequiredConfigurationProperties(config)
                    .quiet()
                    .requireExplicitBindings(false)
                    .setOptionalConfigurationProperties(optionalConfig).initialize();

            KafkaClientConfig clientConfig = KafkaClientConfig.INSTANCE;
            KafkaPluginConfig pluginConfig = KafkaPluginConfig.INSTANCE;
            KafkaConnectorId kafkaConnectorId = KafkaConnectorId.INSTANCE;
            KafkaHiveClient hiveClient = new KafkaHiveClient(kafkaConnectorId,
                clientConfig, pluginConfig);
            KafkaMetadata kafkaMetadata = new KafkaMetadata(hiveClient, kafkaConnectorId);
            KafkaSplitManager kafkaSplitManager = new KafkaSplitManager(hiveClient, kafkaConnectorId, clientConfig);
            KafkaRecordSetProvider kafkaRecordSetProvider = new KafkaRecordSetProvider(kafkaConnectorId);
            KafkaHandleResolver kafkaHandleResolver = new KafkaHandleResolver(kafkaConnectorId);

            ConnectorMetadata connMetadata = new ClassLoaderSafeConnectorMetadata(kafkaMetadata, classLoader);
            ConnectorSplitManager connSplitManager = new ClassLoaderSafeConnectorSplitManager(kafkaSplitManager, classLoader);
            ConnectorRecordSetProvider connRecordSetProvider = new ClassLoaderSafeConnectorRecordSetProvider(kafkaRecordSetProvider, classLoader);
            ConnectorHandleResolver connHandleResolver = new ClassLoaderSafeConnectorHandleResolver(kafkaHandleResolver, classLoader);

            return new KafkaConnector(connMetadata, connSplitManager,
                connRecordSetProvider, connHandleResolver);
            
//            return injector.getInstance(KafkaConnector.class);

//            KafkaMetadata kafkaMetadata = injector.getInstance(KafkaMetadata.class);
//            KafkaSplitManager kafkaSplitManager = injector.getInstance(KafkaSplitManager.class);
//            KafkaRecordSetProvider kafkaRecordSetProvider = injector.getInstance(KafkaRecordSetProvider.class);
//            KafkaHandleResolver kafkaHandleResolver = injector.getInstance(KafkaHandleResolver.class);
//            return new KafkaConnector(kafkaMetadata, kafkaSplitManager,
//                kafkaRecordSetProvider, kafkaHandleResolver);
//            return new KafkaConnector(
//                    new ClassLoaderSafeConnectorMetadata(kafkaMetadata, classLoader),
//                    new ClassLoaderSafeConnectorSplitManager(kafkaSplitManager, classLoader),
//                    new ClassLoaderSafeConnectorRecordSetProvider(kafkaRecordSetProvider, classLoader),
//                    new ClassLoaderSafeConnectorHandleResolver(kafkaHandleResolver, classLoader));
        } catch (Exception e) {
            e.printStackTrace();
            throw Throwables.propagate(e);
        }
    }