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	Description
static class	LicenseStore.ConsumeCallback
static class	LicenseStore.LicenseKeySerde
static class	LicenseStore.LicenseMessageSerde

Field Summary

Fields

Modifier and Type	Field	Description
static final String	MIN_INSYNC_REPLICAS_CONFIG
static final long	READ_TO_END_TIMEOUT_MS
static final String	REPLICATION_FACTOR_CONFIG

Constructor Summary

Constructors

Constructor	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(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)
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)

Method Summary

Modifier and Type	Method	Description
Map<License.DeploymentScope, String>	getInMemoryLicenses()	Returns the current in-memory license state without any topic I/O.
Map<License.DeploymentScope, String>	getLatestLicenses()	Reads the license store LicenseStore and returns the latest licenses per License.DeploymentScope.
boolean	isEventBasedPropagationSupported()	Whether this store supports event-based license propagation.
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	setOnDataChangeListener(Runnable listener)	Set a listener that is called when new license data arrives in the store.
void	start()
void	stop()
String	topic()	LicenseStore falls back to _confluent-command topic if the topic given in the constructor doesn't exist, so this method provides the actual topic used by the license store.

Methods inherited from class Object

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

Field Details

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

Constructor Details

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

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 Details

start

public void start()

Specified by:

start in interface ConfluentLicenseStore

stop

public void stop()

Specified by:

stop in interface ConfluentLicenseStore

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

isEventBasedPropagationSupported

public boolean isEventBasedPropagationSupported()

Description copied from interface: ConfluentLicenseStore

Whether this store supports event-based license propagation. When true, the store can notify listeners immediately when new license records arrive, instead of requiring periodic polling.

Specified by:

isEventBasedPropagationSupported in interface ConfluentLicenseStore

Returns:

true if event-based propagation is supported

setOnDataChangeListener

public void setOnDataChangeListener(Runnable listener)

Description copied from interface: ConfluentLicenseStore

Set a listener that is called when new license data arrives in the store. Only supported when ConfluentLicenseStore.isEventBasedPropagationSupported() returns true.

Specified by:

setOnDataChangeListener in interface ConfluentLicenseStore

Parameters:

listener - the callback to invoke on data change, or null to clear

getInMemoryLicenses

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

Returns the current in-memory license state without any topic I/O. Unlike getLatestLicenses() which calls readToEnd() to ensure the consumer has caught up, this method reads directly from the in-memory latestLicenses map, which is kept up to date by the background WorkThread continuously polling the topic.

Used by the event-based propagation path where the data has just been written to the map by LicenseStore.ConsumeCallback and blocking on readToEnd() would be unnecessary.

Specified by:

getInMemoryLicenses in interface ConfluentLicenseStore

Returns:

map of deployment scope to license JWT, possibly empty

topic

public String topic()

Description copied from interface: ConfluentLicenseStore

LicenseStore falls back to _confluent-command topic if the topic given in the constructor doesn't exist, so this method provides the actual topic used by the license store.

Specified by:

topic in interface ConfluentLicenseStore