code for Kafka and AEM article in medium/Globant

KafkaConfigProperties.java

import org.osgi.service.metatype.annotations.AttributeDefinition;
import org.osgi.service.metatype.annotations.AttributeType;
import org.osgi.service.metatype.annotations.ObjectClassDefinition;

import static com.globant.demos.core.config.KafkaConfig.DEFAULT_AUTO_OFFSET_RESET_CONFIG;
import static com.globant.demos.core.config.KafkaConfig.DEFAULT_CLUSTER_SERVERS;

@ObjectClassDefinition(name = "Kafka Cluster Configuration Properties")
public @interface KafkaConfigProperties {

    @AttributeDefinition(
            name = "Enabled",
            description = "When disabled, the kafka messages will be forever ignored.",
            type = AttributeType.BOOLEAN)
    boolean excalibur_kafka_enabled() default true;

    @AttributeDefinition(
            name = "Enable auto commit",
            description = "If true the consumer's offset will be periodically " +
                    "committed in the background.",
            type = AttributeType.BOOLEAN)
    boolean excalibur_kafka_enable_auto_commit() default false;

    @AttributeDefinition(
            name = "Kafka Consumer Poll Timeout",
            description = "Timeout in millis for the server messages poll",
            type = AttributeType.LONG)
    long excalibur_kafka_poll_timeout() default 2000L;

    @AttributeDefinition(
            name = "Auto offset reset",
            description = "What to do when there is no initial offset in " +
                    "Kafka or if the current offset does not exist any more " +
                    "on the server (e.g. earliest, latest, none)")
    String excalibur_kafka_auto_offset_reset() default DEFAULT_AUTO_OFFSET_RESET_CONFIG;

    @AttributeDefinition(
            name = "Server",
            description = "Comma separated Kafka Cluster Servers, in format of "
                    + "'ip_hostname_1:port,ip_hostname_2:port,ip_hostname_3:port'.")
    String excalibur_kafka_cluster_servers() default DEFAULT_CLUSTER_SERVERS;

    @AttributeDefinition(
            name = "Topics",
            description = "Topic names which AEM Kafka consumer subscribes")
    String [] excalibur_kafka_topics() default {"topic_test"};
}