/**
* An interface for publishing and downloading serialised data to/from payload store.
* The store config will passed down from the original config of the Kafka producer/consumer client. 
*/
public interface PayloadStore implements Configurable, Closeable, Monitorable {
     /**
     * Publish data into the store.
     *
     * @param data data that will be published to the store.
     * @return full path to object in the store.
     * @throw PayloadStoreException in case failed to publish to the store.
     */
     String publish(String topic, byte[] data) throw PayloadStoreException;

    /**
     * Download full data from the store.
     *
     * @param fullPath of the data's reference in the store for example `remote_store/topic_name/<record_random_uuid>`
     * @return content of the object as bytes.
     * @throw PayloadStoreException 
     */
     byte[] download(String fullPath) throw PayloadStoreException;

    /** 
    * Generate an id for the data's reference in the store (Not the full path in the store). 
    * By default the id is a random UUID however some stores might need more smarter way to calculate 
    * its reference id based on the data itself. In such a case please override this method. 
    * @param data data that will be published to the store.
    * @return object id foras example `record_random_uuid`
	*/
    default String id(byte[] data) {
       return UUID.randomUUID().toString();
    }
}

string, the default implementation support UTF-8 encoding. If any other encoding is required the 
    *        implementation will need to address this.
    default String id(byte[] data) {
       return java.util.UUID.randomUUID().toString(); // Java UUID is already composed entirely of ASCII characters which are valid UTF-8
    }
}

Map<String, Object> producerConfigconsumerConfig = new HashMap<>();
producerConfig.put("value.serializersdeserializers", "kafka.serializers.KafkaAvroDeserializer, org.apache.kafka.common.serialization.largemessage.Serializer, kafka.serializers.KafkaAvroDeserializer");
producerConfigconsumerConfig.put("key.serializerdeserializer", "org.apache.kafka.common.serialization.StringSerializer"); producerConfigconsumerConfig.put("large.message.payload.store.class", "myclient.serializerspayloadstore.payload.store.CustomS3Store")
producerConfigconsumerConfig.put("large.message.threshold.bytes", 1048576);
producerConfig.put("s3.bucket", "my-bucket")
producerConfigconsumerConfig.put("s3.retry.attempts", "3");
producerConfigconsumerConfig.put("s3.connection.timeout.ms", "5000");
producerConfigconsumerConfig.put("bootstrap.servers", "localhost:9092");

KafkaProducer<StringKafkaConsumer<String, Double>Object> producerconsumer = new KafkaProducer<>KafkaConsumer<>(producerConfigconsumerConfig);