code: /*******************************************************************************
   * Copyright 2018 572682
   * 
   * 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 us.dot.its.jpo.ode;
  
  import java.util.Properties;
  
  import us.dot.its.jpo.ode.util.SerializableObjectPool;
  import us.dot.its.jpo.ode.wrapper.MessageProducer;
  
  public class SerializableMessageProducerPool<K, V> extends SerializableObjectPool<MessageProducer<K, V>> {
  
     private static final long serialVersionUID = -2293786403623236678L;
  
     transient OdeProperties odeProperties;
  
     private String brokers;
     private String type;
     private String partitionerClass;
  
     private Properties props;
  
     public SerializableMessageProducerPool(OdeProperties odeProperties) {
        super();
        this.odeProperties = odeProperties;
        this.brokers = odeProperties.getKafkaBrokers();
        this.type = odeProperties.getKafkaProducerType();
        this.partitionerClass = odeProperties.getProperty("kafka.partitionerClass");
        init();
     }
  
     public SerializableMessageProducerPool<K, V> init() {
        props = new Properties();
        props.put("acks", odeProperties.getProperty("kafka.producer.ack", MessageProducer.DEFAULT_PRODUCER_ACKS)); // Set
                                                                                                                   // acknowledgments
                                                                                                                   // for
                                                                                                                   // producer
                                                                                                                   // requests.
        props.put("retries",
              odeProperties.getProperty("kafka.producer.retries", MessageProducer.DEFAULT_PRODUCER_RETRIES)); // If
                                                                                                              // the
                                                                                                              // request
                                                                                                              // fails,
                                                                                                              // the
                                                                                                              // producer
                                                                                                              // can
        // automatically retry
        props.put("batch.size",
              odeProperties.getProperty("kafka.producer.batch.size", MessageProducer.DEFAULT_PRODUCER_BATCH_SIZE_BYTES));
        props.put("linger.ms",
              odeProperties.getProperty("kafka.producer.linger.ms", MessageProducer.DEFAULT_PRODUCER_LINGER_MS));
        props.put("buffer.memory", odeProperties.getProperty("kafka.producer.buffer.memory",
              MessageProducer.DEFAULT_PRODUCER_BUFFER_MEMORY_BYTES));
        props.put("key.serializer",
              odeProperties.getProperty("kafka.key.serializer", MessageProducer.SERIALIZATION_STRING_SERIALIZER));
        props.put("value.serializer",
              odeProperties.getProperty("kafka.value.serializer", MessageProducer.SERIALIZATION_BYTE_ARRAY_SERIALIZER));
  
        return this;
     }
  
     @Override
     protected MessageProducer<K, V> create() {
        return new MessageProducer<>(brokers, type, partitionerClass, props,
              odeProperties.getKafkaTopicsDisabledSet());
     }
  
     @Override
     public boolean validate(MessageProducer<K, V> o) {
        return o.getProducer() != null;
     }
  
     @Override
     public void expire(MessageProducer<K, V> o) {
        o.close();
     }
  
     public String getBrokers() {
        return brokers;
     }
  
     public void setBrokers(String brokers) {
        this.brokers = brokers;
     }
  
     public String getType() {
        return type;
     }
  
     public void setType(String type) {
        this.type = type;
     }
  
     public Properties getProps() {
        return props;
     }
  
     public void setProps(Properties props) {
        this.props = props;
     }
  
     public String getPartitionerClass() {
        return partitionerClass;
     }
  
     public void setPartitionerClass(String partitionerClass) {
        this.partitionerClass = partitionerClass;
     }
  
  }
  