Fixes during the workshop

manifest.yaml

@@ -1,9 +1,25 @@
----
 applications:
   - name: spring-kafka-avro
-    path: target/kafka-avro-0.0.1-SNAPSHOT.jar
-    memory: 2G
     instances: 1
     random-route: true
+    buildpacks:
+      - https://github.com/cloudfoundry/java-buildpack
+    path: target/kafka-avro-0.0.1-SNAPSHOT.jar
+    memory: 1G
+    services:
+      - cp
+    JAVA_OPTS: '-Dloader.main=io.confluent.developer.spring.avro.SpringAvroApplication'
+    JBP_CONFIG_JAVA_MAIN: '{java_main_class: org.springframework.boot.loader.PropertiesLauncher}' 
+  - name: spring-kafka-streams
+    instances: 1
+    no-route: true
+    buildpacks:
+      - https://github.com/cloudfoundry/java-buildpack
+    path: target/kafka-avro-0.0.1-SNAPSHOT.jar
+    memory: 1G
+    services:
+      - cp
     env:
-      SPRING_PROFILES_ACTIVE: paas
+      SPRING_PROFILES_ACTIVE: pas
+      JAVA_OPTS: '-Dloader.main=io.confluent.developer.spring.streams.KafkaStreamsApp'
+      JBP_CONFIG_JAVA_MAIN: '{java_main_class: org.springframework.boot.loader.PropertiesLauncher}'

src/main/java/io/confluent/developer/spring/avro/Consumer.java

@@ -1,6 +1,5 @@
 package io.confluent.developer.spring.avro;
 
-import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Service;
@@ -16,7 +15,7 @@ public class Consumer {
   private String topicName;
 
   @KafkaListener(topics = "users", groupId = "group_id")
-  public void consume(ConsumerRecord<String, User> record) {
-    log.info(String.format("Consumed message -> %s", record.value()));
+  public void consume(User record) {
+    log.info(String.format("Consumed message -> %s", record));
   }
 }

src/main/resources/application-cloud.yaml

@@ -26,6 +26,8 @@ spring:
       auto-offset-reset: earliest
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
       value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
+      properties:
+        specific.avro.reader: true
     producer:
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer

src/main/resources/application-pas.yaml

@@ -19,6 +19,8 @@ spring:
       auto-offset-reset: latest
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
       value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
+      properties:
+        specific.avro.reader: true
     producer:
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer

src/main/resources/application.yaml

@@ -16,6 +16,7 @@ spring:
       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
       value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
       properties:
+        specific.avro.reader: true
         interceptor.classes: io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor
     producer:
       key-serializer: org.apache.kafka.common.serialization.StringSerializer