schroedermatt

repositories {
   maven {
      url "http://packages.confluent.io/maven/"
   }
}

compile 'org.springframework.kafka:spring-kafka’
compile "org.apache.avro:avro”
compile "io.confluent:kafka-avro-serializer
testCompile 'org.springframework.kafka:spring-kafka-test'

schroedermatt

@Configuration
class MockSerdeConfig {
   // KafkaProperties groups all properties prefixed with `spring.kafka`
   private KafkaProperties props
   MockSerdeConfig(KafkaProperties kafkaProperties) {
      props = kafkaProperties
   }

   /**
    * Mock schema registry bean used by Kafka Avro Serde since

schroedermatt

spring:
  kafka:
    # point the bootstrap servers to the running embedded kafka
    bootstrap-servers: ${spring.embedded.kafka.brokers}
    consumer:
      client-id: test-avro-consumer
      group-id: test-avro-group
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
    producer:
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer

schroedermatt

// extension on model from 3rd party library
fun ArtistModel.toStruct() : Struct = Struct(Artist.SCHEMA)
        .put(BaseSchema.HREF_FIELD, this.href)
        .put(BaseSchema.ID_FIELD, this.id)
        .put(BaseSchema.NAME_FIELD, this.name)

// example when creating list of SourceRecords from result set
val results: List<ArtistModel> = client.getResults()
results.map {
  SourceRecord(partition, offset, topic, keySchema, it.id, valueSchema, it.toStruct())

schroedermatt

// kotlin data class (equals/hashcode/copy available)
data class Song (val name: String, val artist: Artist, var length: Int?)
  
// pojo with 1 of the Song properties for brevity (and no equals/hashcode)
public class Song {
  private String name;

  public Song(String name) {
    this.name = name;
  }

schroedermatt

@Configuration
public class TopicAdministrator {
  private final TopicConfigurations configurations;
  private final GenericWebApplicationContext context;

  public TopicAdministrator(TopicConfigurations configurations,	GenericWebApplicationContext genericContext) {
    this.configurations = configurations;
    this.context = genericContext;
  }
  

schroedermatt

> bin/kafka-topics.sh --list --zookeeper localhost:2181
test-topic-1
test-topic-2
test-topic-3

schroedermatt

@Configuration
@ConfigurationProperties(prefix = "kafka")
class TopicConfigurations {
   List<TopicConfiguration> topics;

   static class TopicConfiguration {
      String name;
      Integer numPartitions = 3;
      Short replicationFactor = 1;
      

schroedermatt

connect:
    image: confluentinc/cp-kafka-connect:latest
    volumes:
      - ./build/libs:/usr/share/java/kafka-connect-plugin-name

schroedermatt

buildscript {
    dependencies {
        classpath "org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlin_version"
        classpath "com.github.jengelman.gradle.plugins:shadow:$shadow_version"
    }
}

plugins {
    id 'idea'
    id 'com.palantir.git-version' version '0.5.2'