io.confluent.license

Class LicenseStore

java.lang.Object

io.confluent.license.LicenseStore

All Implemented Interfaces:

ConfluentLicenseStore

public class LicenseStore
extends Object
implements ConfluentLicenseStore

IMPORTANT: This class should not use Kafka APIs newer than AK 0.11.0/CP 3.3.0. All of our proprietary connectors are dependent upon the LicenseManager framework and they all include this dependency in their plugin directory. So, when the connectors are installed into any CP version, the connector's LicenseManager uses the AK APIs provided by the Connect installation to create the license topic if necessary. Connectors may be installed into CP 3.3.0 or newer when `AdminClient.createTopics(...)` was added. We should maintain compatibility for the LicenseManager for all these versions by using only older APIs that are available from 3.3.0.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	LicenseStore.ConsumeCallback
static class	LicenseStore.LicenseKeySerde
static class	LicenseStore.LicenseMessageSerde

Field Summary

Fields

Modifier and Type	Field and Description
protected LicenseKafkaBasedLog<io.confluent.command.record.Command.CommandKey,io.confluent.command.record.Command.CommandMessage>	licenseLog
static String	MIN_INSYNC_REPLICAS_CONFIG
static long	READ_TO_END_TIMEOUT_MS
static String	REPLICATION_FACTOR_CONFIG

Constructor Summary

Constructors

Modifier	Constructor and Description
	LicenseStore(io.confluent.command.record.Command.CommandKey commandKey, String topic, ConcurrentHashMap<io.confluent.command.record.Command.CommandKey,String> latestLicenses, LicenseKafkaBasedLog<io.confluent.command.record.Command.CommandKey,io.confluent.command.record.Command.CommandMessage> licenseLog, org.apache.kafka.common.utils.Time time)
	LicenseStore(List<io.confluent.command.record.Command.CommandKey> commandKeys, String topic, ConcurrentHashMap<io.confluent.command.record.Command.CommandKey,String> latestLicenses, LicenseKafkaBasedLog<io.confluent.command.record.Command.CommandKey,io.confluent.command.record.Command.CommandMessage> licenseLog, org.apache.kafka.common.utils.Time time)
	LicenseStore(String topic, Map<String,Object> producerConfig, Map<String,Object> consumerConfig, Map<String,Object> topicConfig)
	LicenseStore(String topic, Map<String,Object> producerConfig, Map<String,Object> consumerConfig, Map<String,Object> topicConfig, Duration topicCreateTimeout, Duration retryBackoffMinMs, Duration retryBackoffMaxMs, org.apache.kafka.common.utils.Time time)
	LicenseStore(String topic, Map<String,Object> producerConfig, Map<String,Object> consumerConfig, Map<String,Object> topicConfig, Duration topicCreateTimeout, Duration retryBackoffMinMs, Duration retryBackoffMaxMs, org.apache.kafka.common.utils.Time time, List<io.confluent.command.record.Command.CommandKey> commandKeys)
	LicenseStore(String topic, Map<String,Object> producerConfig, Map<String,Object> consumerConfig, Map<String,Object> topicConfig, List<io.confluent.command.record.Command.CommandKey> commandKeys)
protected	LicenseStore(String topic, Map<String,Object> producerConfig, Map<String,Object> consumerConfig, Map<String,Object> topicConfig, org.apache.kafka.common.utils.Time time)
protected	LicenseStore(String topic, Map<String,Object> producerConfig, Map<String,Object> consumerConfig, Map<String,Object> topicConfig, org.apache.kafka.common.utils.Time time, List<io.confluent.command.record.Command.CommandKey> commandKeys)

Method Summary

Modifier and Type	Method and Description
protected void	createConfluentLicenseTopic(org.apache.kafka.clients.admin.AdminClient admin, String finalTopic, org.apache.kafka.clients.admin.NewTopic topicDescription)
Map<License.DeploymentScope,String>	getLatestLicenses() Reads the license store LicenseStore and returns the latest licenses per License.DeploymentScope.
String	licenseScan() Returns a single license only when LicenseStore is configured with exactly one Command.CommandKey.
void	registerLicense(License.DeploymentScope deploymentScope, String license) Registers a license JWT for the given License.DeploymentScope.
void	registerLicense(String license) Supports single-license LicenseStore only.
void	registerLicense(String license, org.apache.kafka.clients.producer.Callback callback) Supports single-license LicenseStore only.
void	start()
protected void	startLog()
void	stop()
protected void	stopLog()
String	topic()
protected void	verifyConfluentLicenseTopicMetadata(org.apache.kafka.clients.consumer.Consumer<io.confluent.command.record.Command.CommandKey,io.confluent.command.record.Command.CommandMessage> consumer)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

REPLICATION_FACTOR_CONFIG

public static final String REPLICATION_FACTOR_CONFIG

See Also:

Constant Field Values

MIN_INSYNC_REPLICAS_CONFIG

public static final String MIN_INSYNC_REPLICAS_CONFIG

See Also:

Constant Field Values

READ_TO_END_TIMEOUT_MS

public static final long READ_TO_END_TIMEOUT_MS

See Also:

Constant Field Values

licenseLog

protected final LicenseKafkaBasedLog<io.confluent.command.record.Command.CommandKey,io.confluent.command.record.Command.CommandMessage> licenseLog

Constructor Detail

LicenseStore

public LicenseStore(String topic,
                    Map<String,Object> producerConfig,
                    Map<String,Object> consumerConfig,
                    Map<String,Object> topicConfig)

LicenseStore

public LicenseStore(String topic,
                    Map<String,Object> producerConfig,
                    Map<String,Object> consumerConfig,
                    Map<String,Object> topicConfig,
                    List<io.confluent.command.record.Command.CommandKey> commandKeys)

LicenseStore

protected LicenseStore(String topic,
                       Map<String,Object> producerConfig,
                       Map<String,Object> consumerConfig,
                       Map<String,Object> topicConfig,
                       org.apache.kafka.common.utils.Time time)

LicenseStore

protected LicenseStore(String topic,
                       Map<String,Object> producerConfig,
                       Map<String,Object> consumerConfig,
                       Map<String,Object> topicConfig,
                       org.apache.kafka.common.utils.Time time,
                       List<io.confluent.command.record.Command.CommandKey> commandKeys)

LicenseStore

public LicenseStore(String topic,
                    Map<String,Object> producerConfig,
                    Map<String,Object> consumerConfig,
                    Map<String,Object> topicConfig,
                    Duration topicCreateTimeout,
                    Duration retryBackoffMinMs,
                    Duration retryBackoffMaxMs,
                    org.apache.kafka.common.utils.Time time)

LicenseStore

public LicenseStore(String topic,
                    Map<String,Object> producerConfig,
                    Map<String,Object> consumerConfig,
                    Map<String,Object> topicConfig,
                    Duration topicCreateTimeout,
                    Duration retryBackoffMinMs,
                    Duration retryBackoffMaxMs,
                    org.apache.kafka.common.utils.Time time,
                    List<io.confluent.command.record.Command.CommandKey> commandKeys)

LicenseStore

public LicenseStore(List<io.confluent.command.record.Command.CommandKey> commandKeys,
                    String topic,
                    ConcurrentHashMap<io.confluent.command.record.Command.CommandKey,String> latestLicenses,
                    LicenseKafkaBasedLog<io.confluent.command.record.Command.CommandKey,io.confluent.command.record.Command.CommandMessage> licenseLog,
                    org.apache.kafka.common.utils.Time time)

LicenseStore

public LicenseStore(io.confluent.command.record.Command.CommandKey commandKey,
                    String topic,
                    ConcurrentHashMap<io.confluent.command.record.Command.CommandKey,String> latestLicenses,
                    LicenseKafkaBasedLog<io.confluent.command.record.Command.CommandKey,io.confluent.command.record.Command.CommandMessage> licenseLog,
                    org.apache.kafka.common.utils.Time time)

Method Detail

start

public void start()

Specified by:

start in interface ConfluentLicenseStore

stop

public void stop()

Specified by:

stop in interface ConfluentLicenseStore

startLog

protected void startLog()

stopLog

protected void stopLog()

licenseScan

public String licenseScan()

Returns a single license only when LicenseStore is configured with exactly one Command.CommandKey. When multiple license types are supported, this method returns null. Use getLatestLicenses() to retrieve all licenses by License.DeploymentScope. For multi-license stores, use getLatestLicenses() instead.

Specified by:

licenseScan in interface ConfluentLicenseStore

getLatestLicenses

public Map<License.DeploymentScope,String> getLatestLicenses()

Reads the license store LicenseStore and returns the latest licenses per License.DeploymentScope. Only keys in RECORDS_KEY_ID_MAPPING are included; others are ignored.

Specified by:

getLatestLicenses in interface ConfluentLicenseStore

Returns:

map of scope to latest license JWT (possibly empty)

Throws:

IllegalStateException - if reading the log fails or times out

registerLicense

public void registerLicense(String license)

Supports single-license LicenseStore only. For multi-license configurations use #registerLicense(DeploymentScope, String). Use #registerLicense(DeploymentScope, String) for multi-license stores.

Specified by:

registerLicense in interface ConfluentLicenseStore

registerLicense

public void registerLicense(String license,
                            org.apache.kafka.clients.producer.Callback callback)

Supports single-license LicenseStore only. Use #registerLicense(DeploymentScope, String) for multi-license configurations.

registerLicense

public void registerLicense(License.DeploymentScope deploymentScope,
                            String license)

Registers a license JWT for the given License.DeploymentScope. Only scopes present in LICENSE_SCOPE_TO_KEY_MAPPING are supported.

Specified by:

registerLicense in interface ConfluentLicenseStore

Parameters:

deploymentScope - target scope (e.g. CP, CP_FOR_CC)

license - JWT to write

createConfluentLicenseTopic

protected final void createConfluentLicenseTopic(org.apache.kafka.clients.admin.AdminClient admin,
                                                 String finalTopic,
                                                 org.apache.kafka.clients.admin.NewTopic topicDescription)

verifyConfluentLicenseTopicMetadata

protected final void verifyConfluentLicenseTopicMetadata(org.apache.kafka.clients.consumer.Consumer<io.confluent.command.record.Command.CommandKey,io.confluent.command.record.Command.CommandMessage> consumer)

topic

public String topic()