String modelDir = "src/main/resources/generatedModels/CNN_inception5h";

Path pathGraph = Paths.get(modelDir, "tensorflow_inception_graph.pb");
byte[] graphDef = Files.readAllBytes(pathGraph);

Path pathModel = Paths.get(modelDir, "imagenet_comp_graph_label_strings.txt");
List<String> labels = Files.readAllLines(pathModel, Charset.forName("UTF-8"));

KStream<String, Object> transformedMessage =
imageInputLines.mapValues(value ->  {

	String imageClassification = "unknown";
	String imageProbability = "unknown";

	String imageFile = value;

	Path pathImage = Paths.get(imageFile);
	byte[] imageBytes;
	try {
		imageBytes = Files.readAllBytes(pathImage);

try (Tensor image = constructAndExecuteGraphToNormalizeImage(imageBytes)) {
	float[] labelProbabilities = executeInceptionGraph(graphDef, image);
	int bestLabelIdx = maxIndex(labelProbabilities);

	imageClassification = labels.get(bestLabelIdx);

	imageProbability = Float.toString(labelProbabilities[bestLabelIdx] * 100f);

	System.out.println(String.format("BEST MATCH: %s (%.2f%% likely)", imageClassification,
			labelProbabilities[bestLabelIdx] * 100f));
}

final Output input = b.constant("input", imageBytes);
final Output output = b
		.div(b.sub(
				b.resizeBilinear(b.expandDims(b.cast(b.decodeJpeg(input, 3), DataType.FLOAT),
						b.constant("make_batch", 0)), b.constant("size", new int[] { H, W })),
				b.constant("mean", mean)), b.constant("scale", scale));
try (Session s = new Session(g)) {
	return s.runner().fetch(output.op().name()).run().get(0);
}

try (Graph g = new Graph()) {

	// Model loading: Using Graph.importGraphDef() to load a pre-trained Inception
	// model.
	g.importGraphDef(graphDef);

	// Graph execution: Using a Session to execute the graphs and find the best
	// label for an image.
	try (Session s = new Session(g);
			Tensor result = s.runner().feed("input", image).fetch("output").run().get(0)) {
		final long[] rshape = result.shape();
		if (result.numDimensions() != 2 || rshape[0] != 1) {
			throw new RuntimeException(String.format(
					"Expected model to produce a [1 N] shaped tensor where N is the number of labels, instead it produced one with shape %s",
					Arrays.toString(rshape)));
		}
		int nlabels = (int) rshape[1];
		return result.copyTo(new float[1][nlabels])[0];
	}
}

@ClassRule
public static final EmbeddedKafkaCluster CLUSTER = new TestEmbeddedKafkaCluster(1);

private static final String inputTopic = "IrisInputTopic";
private static final String outputTopic = "IrisOutputTopic";

// Generated DL4J model
private File locationDL4JModel = new File("src/main/resources/generatedModels/DL4J/DL4J_Iris_Model.zip");

// Prediction Value
private static String irisPrediction = "unknown";

		// Iris input data (the model returns probabilities for input being each of Iris
		// Type 1, 2 and 3)
		List<String> inputValues = Arrays.asList("5.4,3.9,1.7,0.4", "7.0,3.2,4.7,1.4", "4.6,3.4,1.4,0.3");

(snip)

		streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
		streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

// Create DL4J object (see DeepLearning4J_CSV_Model.java)
MultiLayerNetwork model = ModelSerializer.restoreMultiLayerNetwork(locationDL4JModel);

final StreamsBuilder builder = new StreamsBuilder();
final KStream<String, String> irisInputLines = builder.stream(inputTopic);

irisInputLines.foreach((key, value) -> {

	if (value != null && !value.equals("")) {
		System.out.println("#####################");
		System.out.println("Iris Input:" + value);

		// TODO Easier way to map from String[] to double[] !!!
		String[] stringArray = value.split(",");
		Double[] doubleArray = Arrays.stream(stringArray).map(Double::valueOf).toArray(Double[]::new);
		double[] irisInput = Stream.of(doubleArray).mapToDouble(Double::doubleValue).toArray();

		// Inference
		INDArray input = Nd4j.create(irisInput);
		INDArray result = model.output(input);

		System.out.println("Probabilities: " + result.toString());

		irisPrediction = result.toString();

	}

});

// Transform message: Add prediction information
KStream<String, Object> transformedMessage = irisInputLines
		.mapValues(value -> "Prediction: Iris Probability => " + irisPrediction);

// Send prediction information to Output Topic
transformedMessage.to(outputTopic);

final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);
streams.cleanUp();
streams.start();
System.out.println("Iris Prediction Microservice is running...");
System.out.println("Input to Kafka Topic 'IrisInputTopic'; Output to Kafka Topic 'IrisOutputTopic'");

Properties producerConfig = new Properties();
producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
producerConfig.put(ProducerConfig.RETRIES_CONFIG, 0);
producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
IntegrationTestUtils.produceValuesSynchronously(inputTopic, inputValues, producerConfig, new MockTime());

public void shouldPredictValues() throws Exception {

String simpleMlp = new ClassPathResource("generatedModels/Keras/simple_mlp.h5").getFile().getPath();
System.out.println(simpleMlp.toString());

MultiLayerNetwork model = KerasModelImport.importKerasSequentialModelAndWeights(simpleMlp);

Properties streamsConfiguration = new Properties();
streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG,
		"kafka-streams-tensorflow-keras-integration-test");
streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());

streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

final StreamsBuilder builder = new StreamsBuilder();

final KStream<String, String> inputEvents = builder.stream(inputTopic);

inputEvents.foreach((key, value) -> {

	// Transform input values (list of Strings) to expected DL4J parameters (two
	// Integer values):
	String[] valuesAsArray = value.split(",");
	INDArray input = Nd4j.create(Integer.parseInt(valuesAsArray[0]), Integer.parseInt(valuesAsArray[1]));

	output = model.output(input);
	prediction = output.toString();

});

// Transform message: Add prediction result
KStream<String, Object> transformedMessage = inputEvents.mapValues(value -> "Prediction => " + prediction);

// Send prediction result to Output Topic
transformedMessage.to(outputTopic);

// Start Kafka Streams Application to process new incoming messages from
// Input Topic
final KafkaStreams streams = new TestKafkaStreams(builder.build(), streamsConfiguration);
streams.cleanUp();
streams.start();
System.out.println("Prediction Microservice is running...");
System.out.println("Input to Kafka Topic " + inputTopic + "; Output to Kafka Topic " + outputTopic);

Vagrant.configure("2") do |config|
  config.vm.define "docker-01" do |config|
        config.vm.box = "bento/ubuntu-18.04"
        config.vm.network :private_network, ip: "172.16.19.220"
        config.vm.hostname = "docker-01"
  end

  config.vm.provision "shell", inline: <<-SHELL
    apt-get update
    apt-get install -y \
    apt-transport-https \
    ca-certificates \
    curl \
    gnupg-agent \
    software-properties-common
    curl -fsSL https://download.docker.com/linux/ubuntu/gpg | apt-key add -
    apt-key fingerprint 0EBFCD88
    add-apt-repository \
    "deb [arch=amd64] https://download.docker.com/linux/ubuntu \
    $(lsb_release -cs) \
    stable"
    apt-get update
    apt-get install -y docker-ce docker-ce-cli containerd.io
    sed -i "s;fd://;tcp://0.0.0.0:2375;g" /lib/systemd/system/docker.service
  SHELL

end

$ sudo docker --host tcp://172.16.19.220:2375 run hello-world

BrokerInfo(Broker(config.brokerId, updatedEndpoints, config.rack), config.interBrokerProtocolVersion, jmxPort)

commit e70a191d3038e00790aa95fbd1e16e78c32b79a4
Author: Ismael Juma <ismael@juma.me.uk>
Date:   Mon May 21 23:17:42 2018 -0700

    KAFKA-4423: Drop support for Java 7 (KIP-118) and update deps (#5046)

(snip)

diff --git a/docs/ops.html b/docs/ops.html
index 450a268a2..95b9a9601 100644
--- a/docs/ops.html
+++ b/docs/ops.html
@@ -639,9 +639,7 @@

   From a security perspective, we recommend you use the latest released version of JDK 1.8 as older freely available versions have disclosed security vulnerabilities.

-  LinkedIn is currently running JDK 1.8 u5 (looking to upgrade to a newer version) with the G1 collector. If you decide to use the G1 collector (the current default) and you are still on JDK 1.7, make sure you are on u51 or newer. LinkedIn tried out u21 in testing, but they had a number of problems with the GC implementation in that version.
-
-  LinkedIn's tuning looks like this:
+  LinkedIn is currently running JDK 1.8 u5 (looking to upgrade to a newer version) with the G1 collector. LinkedIn's tuning looks like this:
   <pre class="brush: text;">
   -Xmx6g -Xms6g -XX:MetaspaceSize=96m -XX:+UseG1GC
   -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M

/home/dobachi/.local/share/virtualenvs/openml_sklearn_example-YW762zpK/lib/python3.7/site-packages/pandas/compat/__init__.py:85: UserWarning: Could not import the lzma module. Your installed Python is incomplete. Attempting to use lzma compression will result in a RuntimeError.
  warnings.warn(msg)

$ sudo apt install liblzma-dev

$ pyenv install 3.7.5
$ pyenv global 3.7.5
$ pip install pipenv