Package fr.codinbox.connector.commons.kafka

Class KafkaConnectionImpl

java.lang.Object

fr.codinbox.connector.commons.kafka.KafkaConnectionImpl

All Implemented Interfaces:

KafkaConnection

public class KafkaConnectionImpl
extends Object
implements KafkaConnection

Implementation of KafkaConnection that loads Kafka properties from a file, provides factory methods for creating clients, and tracks them for graceful shutdown.

See Also:

• KafkaConnection

Constructor Summary

Constructors

Constructor	Description
KafkaConnectionImpl(@NotNull Logger logger, @NotNull String id, @NotNull String configFilePath)	Creates a new Kafka connection instance.

Method Summary

Modifier and Type	Method	Description
@NotNull AdminClient	createAdminClient()	Creates a new AdminClient using the base configuration properties.
@NotNull AdminClient	createAdminClient(@NotNull Properties overrides)	Creates a new AdminClient with the specified property overrides.
@NotNull KafkaConsumer<String,String>	createConsumer()	Creates a new KafkaConsumer with String key and value deserializers.
<K, V> @NotNull KafkaConsumer<K,V>	createConsumer(@NotNull Properties overrides)	Creates a new KafkaConsumer with the specified property overrides.
@NotNull KafkaProducer<String,String>	createProducer()	Creates a new KafkaProducer with String key and value serializers.
<K, V> @NotNull KafkaProducer<K,V>	createProducer(@NotNull Properties overrides)	Creates a new KafkaProducer with the specified property overrides.
@NotNull Properties	getProperties()	Returns a defensive copy of the loaded Kafka properties.
void	init()	Initializes the connection by loading and validating the configuration properties.
boolean	isExitOnFailure()	Returns whether this connection is configured to cause server shutdown on initialization failure.
void	shutdown()	Shuts down all tracked Kafka clients gracefully with the configured timeout.

Methods inherited from class java.lang.Object

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

Constructor Details

KafkaConnectionImpl

public KafkaConnectionImpl(@NotNull
 @NotNull Logger logger,
 @NotNull
 @NotNull String id,
 @NotNull
 @NotNull String configFilePath)

Creates a new Kafka connection instance.

Parameters:

logger - the logger for diagnostic messages

id - the connection identifier

configFilePath - the path to the .properties configuration file

Method Details

init

public void init()
          throws ConnectionInitException

Description copied from interface: KafkaConnection

Initializes the connection by loading and validating the configuration properties.

Specified by:

init in interface KafkaConnection

Throws:

ConnectionInitException - if the properties file cannot be loaded or bootstrap.servers is missing

getProperties

@NotNull
public @NotNull Properties getProperties()

Description copied from interface: KafkaConnection

Returns a defensive copy of the loaded Kafka properties.

Modifications to the returned Properties object do not affect the internal configuration.

Specified by:

getProperties in interface KafkaConnection

Returns:

a copy of the Kafka configuration properties

createProducer

@NotNull
public @NotNull KafkaProducer<String,String> createProducer()

Description copied from interface: KafkaConnection

Creates a new KafkaProducer with String key and value serializers.

The returned producer is tracked and will be closed on KafkaConnection.shutdown().

Specified by:

createProducer in interface KafkaConnection

Returns:

a new Kafka producer configured with the loaded properties

createProducer

@NotNull
public <K,
V> @NotNull KafkaProducer<K,V> createProducer(@NotNull
 @NotNull Properties overrides)

Description copied from interface: KafkaConnection

Creates a new KafkaProducer with the specified property overrides.

Override properties are merged on top of the base configuration (overrides win). The returned producer is tracked and will be closed on KafkaConnection.shutdown().

Specified by:

createProducer in interface KafkaConnection

Type Parameters:

K - the key type

V - the value type

Parameters:

overrides - additional properties to merge (e.g., custom serializers)

Returns:

a new Kafka producer

createConsumer

@NotNull
public @NotNull KafkaConsumer<String,String> createConsumer()

Description copied from interface: KafkaConnection

Creates a new KafkaConsumer with String key and value deserializers.

The returned consumer is tracked and will be closed on KafkaConnection.shutdown().

Specified by:

createConsumer in interface KafkaConnection

Returns:

a new Kafka consumer configured with the loaded properties

createConsumer

@NotNull
public <K,
V> @NotNull KafkaConsumer<K,V> createConsumer(@NotNull
 @NotNull Properties overrides)

Description copied from interface: KafkaConnection

Creates a new KafkaConsumer with the specified property overrides.

Override properties are merged on top of the base configuration (overrides win). The returned consumer is tracked and will be closed on KafkaConnection.shutdown().

Specified by:

createConsumer in interface KafkaConnection

Type Parameters:

K - the key type

V - the value type

Parameters:

overrides - additional properties to merge (e.g., group.id, custom deserializers)

Returns:

a new Kafka consumer

createAdminClient

@NotNull
public @NotNull AdminClient createAdminClient()

Description copied from interface: KafkaConnection

Creates a new AdminClient using the base configuration properties.

The returned client is tracked and will be closed on KafkaConnection.shutdown().

Specified by:

createAdminClient in interface KafkaConnection

Returns:

a new Kafka admin client

createAdminClient

@NotNull
public @NotNull AdminClient createAdminClient(@NotNull
 @NotNull Properties overrides)

Description copied from interface: KafkaConnection

Creates a new AdminClient with the specified property overrides.

The returned client is tracked and will be closed on KafkaConnection.shutdown().

Specified by:

createAdminClient in interface KafkaConnection

Parameters:

overrides - additional properties to merge

Returns:

a new Kafka admin client

shutdown

public void shutdown()

Description copied from interface: KafkaConnection

Shuts down all tracked Kafka clients gracefully with the configured timeout.

The timeout is controlled by the shutdownTimeoutMs property (default: 30000 milliseconds).

Specified by:

shutdown in interface KafkaConnection

isExitOnFailure

public boolean isExitOnFailure()

Description copied from interface: KafkaConnection

Returns whether this connection is configured to cause server shutdown on initialization failure.

Specified by:

isExitOnFailure in interface KafkaConnection

Returns:

true if the server should exit when this connection fails to initialize