Step 4: Implement the Producer

Refer to the following code breakdown:

• The PutRecord API expects a byte array, and you need to convert trade to JSON format. This single line of code performs that operation:

  byte[] bytes = trade.toJsonAsBytes();

• Before you can send the trade, you create a new PutRecordRequest instance (called putRecord in this case):

  PutRecordRequest putRecord = new PutRecordRequest();

  Each PutRecord call requires the stream name, partition key, and data blob. The following code populates these fields in the putRecord object using its setXxxx() methods:

  putRecord.setStreamName(streamName);
  putRecord.setPartitionKey(trade.getTickerSymbol());
  putRecord.setData(ByteBuffer.wrap(bytes));

  The example uses a stock ticket as a partition key, which maps the record to a specific shard. In practice, you should have hundreds or thousands of partition keys per shard such that records are evenly dispersed across your stream. For more information about how to add data to a stream, see Adding Data to a Stream.

  Now putRecord is ready to send to the client (the put operation):

  kinesisClient.putRecord(putRecord);

• Error checking and logging are always useful additions. This code logs error conditions:

  if (bytes == null) {
      LOG.warn("Could not get JSON bytes for stock trade");
      return;
  }

  Add the try/catch block around the put operation:

  try {
         kinesisClient.putRecord(putRecord);
  } catch (AmazonClientException ex) {
         LOG.warn("Error sending record to Amazon Kinesis.", ex);
  }

  This is because a Kinesis Data Streams put operation can fail because of a network error, or due to the stream reaching its throughput limits and getting throttled. We recommend carefully considering your retry policy for put operations to avoid data loss, such using as a simple retry.

• Status logging is helpful but optional:

  LOG.info("Putting trade: " + trade.toString());