@Override
	public void register(PolarisRegistration registration) {

		if (StringUtils.isBlank(registration.getServiceId())) {
			LOGGER.warn("No service to register for polaris client...");
			return;
		}
		registration.customize();
		String serviceId = registration.getServiceId();

		// Register instance.
		InstanceRegisterRequest instanceRegisterRequest = new InstanceRegisterRequest();
		instanceRegisterRequest.setNamespace(polarisDiscoveryProperties.getNamespace());
		instanceRegisterRequest.setService(serviceId);
		instanceRegisterRequest.setHost(registration.getHost());
		instanceRegisterRequest.setPort(registration.getPort());
		instanceRegisterRequest.setWeight(polarisDiscoveryProperties.getWeight());
		instanceRegisterRequest.setToken(polarisDiscoveryProperties.getToken());
		instanceRegisterRequest.setRegion(staticMetadataManager.getRegion());
		instanceRegisterRequest.setZone(staticMetadataManager.getZone());
		instanceRegisterRequest.setCampus(staticMetadataManager.getCampus());
		instanceRegisterRequest.setTtl(polarisDiscoveryProperties.getHeartbeatInterval());
		instanceRegisterRequest.setMetadata(registration.getMetadata());
		instanceRegisterRequest.setExtendedMetadata(registration.getExtendedMetadata());
		instanceRegisterRequest.setProtocol(polarisDiscoveryProperties.getProtocol());
		instanceRegisterRequest.setVersion(polarisDiscoveryProperties.getVersion());
		instanceRegisterRequest.setInstanceId(polarisDiscoveryProperties.getInstanceId());
		CalleeMetadataContainerGroup.getStaticApplicationMetadataContainer()
				.putMetadataStringValue(MetadataConstants.LOCAL_NAMESPACE, polarisDiscoveryProperties.getNamespace(), TransitiveType.DISPOSABLE);
		CalleeMetadataContainerGroup.getStaticApplicationMetadataContainer()
				.putMetadataStringValue(MetadataConstants.LOCAL_SERVICE, serviceId, TransitiveType.DISPOSABLE);
		CalleeMetadataContainerGroup.getStaticApplicationMetadataContainer()
				.putMetadataStringValue(MetadataConstants.LOCAL_IP, registration.getHost(), TransitiveType.DISPOSABLE);
		CalleeMetadataContainerGroup.getStaticApplicationMetadataContainer()
				.putMetadataStringValue(MetadataConstants.LOCAL_PORT, String.valueOf(registration.getPort()), TransitiveType.DISPOSABLE);
		try {
			ProviderAPI providerClient = polarisSDKContextManager.getProviderAPI();
			InstanceRegisterResponse instanceRegisterResponse;
			if (StringUtils.isBlank(polarisDiscoveryProperties.getHealthCheckUrl())) {
				instanceRegisterResponse = providerClient.registerInstance(instanceRegisterRequest);
			}
			else {
				instanceRegisterResponse = providerClient.register(instanceRegisterRequest);
				InstanceHeartbeatRequest heartbeatRequest = new InstanceHeartbeatRequest();
				BeanUtils.copyProperties(instanceRegisterRequest, heartbeatRequest);
				// Start the heartbeat thread after the registration is successful.
				heartbeat(heartbeatRequest);
			}
			registration.setInstanceId(instanceRegisterResponse.getInstanceId());
			LOGGER.info("polaris registry, {} {} {} {}:{} {} {} {} {} register finished", polarisDiscoveryProperties.getNamespace(),
					registration.getServiceId(), registration.getInstanceId(), registration.getHost(), registration.getPort(),
					staticMetadataManager.getRegion(), staticMetadataManager.getZone(), staticMetadataManager.getCampus(),
					staticMetadataManager.getMergedStaticMetadata());
			if (Objects.nonNull(polarisStatProperties) && polarisStatProperties.isEnabled()) {
				try {
					StatReporter statReporter = (StatReporter) polarisSDKContextManager.getSDKContext().getPlugins()
							.getPlugin(PluginTypes.STAT_REPORTER.getBaseType(), StatReporterConfig.DEFAULT_REPORTER_PROMETHEUS);
					if (Objects.nonNull(statReporter)) {
						ReporterMetaInfo reporterMetaInfo = statReporter.metaInfo();
						if (reporterMetaInfo.getPort() != null) {
							LOGGER.info("Stat server started on port: " + reporterMetaInfo.getPort() + " (http)");
						}
						else {
							LOGGER.info("Stat server is set to type of Push gateway");
						}
					}
					else {
						LOGGER.warn("Plugin StatReporter not found");
					}
				}
				catch (Exception e) {
					LOGGER.warn("Stat server started error, ", e);
				}
			}

			ServiceConfigImpl serviceConfig = (ServiceConfigImpl) polarisSDKContextManager.getSDKContext().getConfig()
					.getProvider().getService();
			serviceConfig.setNamespace(polarisDiscoveryProperties.getNamespace());
			serviceConfig.setName(serviceId);

			PolarisSDKContextManager.isRegistered = true;
		}
		catch (Exception e) {
			LOGGER.error("polaris registry, {} register failed...{},", registration.getServiceId(), registration, e);
			rethrowRuntimeException(e);
		}
	}
@Test
	public void testRegister() {
		this.contextRunner.run(context -> {
			PolarisServiceRegistry registry = context.getBean(PolarisServiceRegistry.class);
			PolarisRegistration registration = Mockito.mock(PolarisRegistration.class);
			when(registration.getHost()).thenReturn("127.0.0.1");
			when(registration.getPort()).thenReturn(PORT);
			when(registration.getServiceId()).thenReturn(SERVICE_PROVIDER);

			assertThatCode(() -> {
				registry.register(registration);
			}).doesNotThrowAnyException();

			assertThatCode(() -> {
				assertThat(registry.getStatus(registration)).isEqualTo("DOWN");
			}).doesNotThrowAnyException();

			assertThatCode(() -> {
				registry.deregister(registration);
			}).doesNotThrowAnyException();
		});
	}