Implementar el productor

private static void sendStockTrade(StockTrade trade, KinesisAsyncClient kinesisClient,
            String streamName) {
        byte[] bytes = trade.toJsonAsBytes();
        // The bytes could be null if there is an issue with the JSON serialization by the Jackson JSON library.
        if (bytes == null) {
            LOG.warn("Could not get JSON bytes for stock trade");
            return;
        }

        LOG.info("Putting trade: " + trade.toString());
        PutRecordRequest request = PutRecordRequest.builder()
                .partitionKey(trade.getTickerSymbol()) // We use the ticker symbol as the partition key, explained in the Supplemental Information section below.
                .streamName(streamName)
                .data(SdkBytes.fromByteArray(bytes))
                .build();
        try {
            kinesisClient.putRecord(request).get();
        } catch (InterruptedException e) {
            LOG.info("Interrupted, assuming shutdown.");
        } catch (ExecutionException e) {
            LOG.error("Exception while sending data to Kinesis. Will try again next cycle.", e);
        }
    }
                    

Consulte el siguiente desglose del código:

• La API de PutRecord espera una matriz de bytes y debe convertir la transacción al formato JSON. Esta única línea de código realiza esa operación:

  
  byte[] bytes = trade.toJsonAsBytes();
                              

• Antes de poder enviar la transacción, debe crear una nueva instancia de PutRecordRequest (denominada solicitud en este caso). Cada llamada a request requiere el nombre de la secuencia, la clave de partición y el blob de datos.

  
  PutPutRecordRequest request = PutRecordRequest.builder()
      .partitionKey(trade.getTickerSymbol()) // We use the ticker symbol as the partition key, explained in the Supplemental Information section below.
      .streamName(streamName)
      .data(SdkBytes.fromByteArray(bytes))
      .build();
                              

  El ejemplo utiliza una graduación de existencias como clave de partición, que asigna el registro a una partición específica. En la práctica, debería tener cientos o miles de claves de partición por fragmento, de forma que los registros se dispersen de forma uniforme en toda la secuencia. Para obtener más información acerca de cómo agregar datos a una secuencia, consulte Datos de escritura en Amazon Kinesis Data Streams.

  Ahora request estará listo para el envío al cliente (la operación PUT):

  
     kinesisClient.putRecord(request).get();
                              

• Siempre es útil agregar funciones de comprobación y registro de errores. Este código registra las condiciones de error:

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

  Agregue el bloque try/catch a la operación put:

  
  try {
   	kinesisClient.putRecord(request).get();
  } catch (InterruptedException e) {
              LOG.info("Interrupted, assuming shutdown.");
  } catch (ExecutionException e) {
              LOG.error("Exception while sending data to Kinesis. Will try again next cycle.", e);
  }
                              

  Esto se debe a que una operación put de Kinesis Data Streams puede fallar debido a un error de red o debido a que la secuencia de datos alcanza sus límites de rendimiento y se estrangula. Se recomienda comprobar detalladamente su política de reintentos para las operaciones de put de modo que evite la pérdida de datos, por ejemplo, utilizando un reintento.

• El registro de estado resulta útil, pero es opcional:

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