// Path: common/src/main/java/com/khs/microservice/whirlpool/common/Message.java
// public class Message {
//    private String type;
//    private String id;
// 
//    public String getType() { return type; }
// 
//    public void setType(String type) { this.type = type; }
// 
//    public String getId() { return id; }
// 
//    public void setId(String id) { this.id = id; }
// 
//    @Override
//    public boolean equals(Object o) {
//       if (this == o) { return true; }
//       if (o == null || getClass() != o.getClass()) { return false; }
// 
//       Message that = (Message) o;
// 
//       if (type != null ? !type.equals(that.type) : that.type != null) { return false; }
//       if (id != null ? !id.equals(that.id) : that.id != null) { return false; }
// 
//       return true;
//    }
// 
//    @Override
//    public int hashCode() {
//       int result = type != null ? type.hashCode() : 0;
//       result = 31 * result + (this.id != null ? this.id.hashCode() : 0);
//       return result;
//    }
// 
//    @Override
//    public String toString() {
//       return "Message{" +
//               "type='"  + type + "'" +
//               ", id='"  + id   + "'" +
//              "}";
//    }
// }


import com.google.common.io.Resources;
import com.google.gson.Gson;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.InputStream;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import com.khs.microservice.whirlpool.common.Message;
package com.khs.microservice.whirlpool.whirlpoolserver;






/**
  * This class waits for command requests from the client request queue,
  * then sends them to Kafka
  *
  * @author jwb
  */
public class WhirlpoolKafkaProducer implements Callable<String> {
    private static final Logger logger = LoggerFactory.getLogger(WhirlpoolKafkaProducer.class);
    private static final AtomicBoolean keepRunning = new AtomicBoolean(true);
    private ConcurrentLinkedQueue<String> requestQueue;
    private Gson gson = new Gson();

    WhirlpoolKafkaProducer(ConcurrentLinkedQueue<String> requestQueue) {
        this.requestQueue = requestQueue;
    }

    @Override
    public String call() throws Exception {
        // set up the producer
        KafkaProducer<String, String> producer;
        try (InputStream props = Resources.getResource("producer.props").openStream()) {
            Properties properties = new Properties();
            properties.load(props);
            producer = new KafkaProducer<>(properties);
        }

        try {
            String request;

            while (keepRunning.get()) {
                while ((request = requestQueue.poll()) != null) {
                    // simple class containing only the type
// Path: common/src/main/java/com/khs/microservice/whirlpool/common/Message.java
// public class Message {
//    private String type;
//    private String id;
// 
//    public String getType() { return type; }
// 
//    public void setType(String type) { this.type = type; }
// 
//    public String getId() { return id; }
// 
//    public void setId(String id) { this.id = id; }
// 
//    @Override
//    public boolean equals(Object o) {
//       if (this == o) { return true; }
//       if (o == null || getClass() != o.getClass()) { return false; }
// 
//       Message that = (Message) o;
// 
//       if (type != null ? !type.equals(that.type) : that.type != null) { return false; }
//       if (id != null ? !id.equals(that.id) : that.id != null) { return false; }
// 
//       return true;
//    }
// 
//    @Override
//    public int hashCode() {
//       int result = type != null ? type.hashCode() : 0;
//       result = 31 * result + (this.id != null ? this.id.hashCode() : 0);
//       return result;
//    }
// 
//    @Override
//    public String toString() {
//       return "Message{" +
//               "type='"  + type + "'" +
//               ", id='"  + id   + "'" +
//              "}";
//    }
// }


// Path: whirlpoolserver/src/main/java/com/khs/microservice/whirlpool/whirlpoolserver/WhirlpoolKafkaProducer.java
import com.google.common.io.Resources;
import com.google.gson.Gson;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.InputStream;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import com.khs.microservice.whirlpool.common.Message;

package com.khs.microservice.whirlpool.whirlpoolserver;






/**
  * This class waits for command requests from the client request queue,
  * then sends them to Kafka
  *
  * @author jwb
  */
public class WhirlpoolKafkaProducer implements Callable<String> {
    private static final Logger logger = LoggerFactory.getLogger(WhirlpoolKafkaProducer.class);
    private static final AtomicBoolean keepRunning = new AtomicBoolean(true);
    private ConcurrentLinkedQueue<String> requestQueue;
    private Gson gson = new Gson();

    WhirlpoolKafkaProducer(ConcurrentLinkedQueue<String> requestQueue) {
        this.requestQueue = requestQueue;
    }

    @Override
    public String call() throws Exception {
        // set up the producer
        KafkaProducer<String, String> producer;
        try (InputStream props = Resources.getResource("producer.props").openStream()) {
            Properties properties = new Properties();
            properties.load(props);
            producer = new KafkaProducer<>(properties);
        }

        try {
            String request;

            while (keepRunning.get()) {
                while ((request = requestQueue.poll()) != null) {
                    // simple class containing only the type
                    Message message = gson.fromJson(request, Message.class);