Kafka Transactional Sample

- Added transactional and some more required properties
 - Setting read_commited in consumer
 - Added license and author tag
 - Added more fleshed out readme
 - Added alternate failure and removed unique index
 - Update spring boot starter parent to 2.1.0.BUILD-SNAPSHOT

transaction-kafka-samples/transaction-spring-data-processor/src/main/java/sample/processor/Person.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package sample.processor;
+
+import java.util.Objects;
+import java.util.StringJoiner;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
+import javax.persistence.Id;
+import javax.persistence.Table;
+import javax.persistence.UniqueConstraint;
+import javax.validation.constraints.Max;
+
+import org.hibernate.validator.constraints.Length;
+
+/**
+ * @author Bjarte Stien Karlsen
+ */
+@Entity
+@Table(name = "person")
+class Person {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    private String name;
+
+    public Long getId() {
+        return id;
+    }
+
+    public void setId(Long id) {
+        this.id = id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    @Override
+    public String toString() {
+        return new StringJoiner(", ", Person.class.getSimpleName() + "[", "]")
+            .add("id=" + id)
+            .add("name='" + name + "'")
+            .toString();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        Person person = (Person) o;
+        return Objects.equals(id, person.id) &&
+            Objects.equals(name, person.name);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(id, name);
+    }
+}

transaction-kafka-samples/transaction-spring-data-processor/src/main/java/sample/processor/PersonRepository.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package sample.processor;
+
+import org.springframework.data.repository.CrudRepository;
+
+/**
+ * @author Bjarte Stien Karlsen
+ */
+interface PersonRepository extends CrudRepository<Person, String> {
+
+}

transaction-kafka-samples/transaction-spring-data-processor/src/main/java/sample/processor/ProcessorApplication.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package sample.processor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.messaging.handler.annotation.SendTo;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
+
+import javax.transaction.Transactional;
+import java.util.StringJoiner;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * @author Bjarte Stien Karlsen
+ */
+@SpringBootApplication
+@EnableBinding(Processor.class)
+@EnableTransactionManagement
+public class ProcessorApplication {
+
+    private Logger logger = LoggerFactory.getLogger(this.getClass().getName());
+
+    private AtomicBoolean shouldFail= new AtomicBoolean(false);
+    private PersonRepository repository;
+
+    public ProcessorApplication(PersonRepository repository) {
+
+        this.repository = repository;
+    }
+
+    public static void main(String[] args) {
+        SpringApplication.run(ProcessorApplication.class, args);
+    }
+
+    @Transactional
+    @StreamListener(Processor.INPUT)
+    @SendTo(Processor.OUTPUT)
+    public PersonEvent process(PersonEvent data) {
+        logger.info("Received event={}", data);
+        Person person = new Person();
+        person.setName(data.getName());
+
+        if(shouldFail.get()) {
+            shouldFail.set(false);
+            throw new RuntimeException("Simulated network error");
+        } else {
+            //We fail every other request as a test
+            shouldFail.set(true);
+        }
+        logger.info("Saving person={}", person);
+
+        Person savedPerson = repository.save(person);
+
+        PersonEvent event = new PersonEvent();
+        event.setName(savedPerson.getName());
+        event.setType("PersonSaved");
+        logger.info("Sent event={}", event);
+        return event;
+    }
+
+    static class PersonEvent {
+
+        String name;
+        String type;
+
+        public String getName() {
+            return name;
+        }
+
+        public void setName(String name) {
+            this.name = name;
+        }
+
+        public String getType() {
+            return type;
+        }
+
+        public void setType(String type) {
+            this.type = type;
+        }
+
+        @Override
+        public String toString() {
+            return new StringJoiner(", ", PersonEvent.class.getSimpleName() + "[", "]")
+                .add("name='" + name + "'")
+                .add("type='" + type + "'")
+                .toString();
+        }
+    }
+}

transaction-kafka-samples/transaction-spring-data-processor/src/main/resources/application.yml

@@ -0,0 +1,19 @@
+spring:
+    cloud.stream:
+        bindings:
+            input.destination: person-command
+            output.destination: person-event
+        kafka.binder.transaction:
+            transaction-id-prefix: person-
+            producer:
+              configuration:
+                retries: 1
+                acks: all
+
+    datasource:
+        url: jdbc:mariadb://localhost:3306/sample_mysql_db
+        username: root
+        password: pwd
+        driver-class-name: org.mariadb.jdbc.Driver
+    jpa.hibernate.ddl-auto: create-drop
+server.port: 9998