List<KinesisLogEntry> processMessages(final byte[] payloadBytes, Instant approximateArrivalTimestamp) throws IOException {

        // This method will be called from a codec, and therefore will not perform any detection. It will rely
        // exclusively on the AWSMessageType detected in the setup HealthCheck.
        // If a user needs to change the type of data stored in a stream, they will need to set the integration up again.
        if (awsMessageType == AWSMessageType.KINESIS_CLOUDWATCH_FLOW_LOGS || awsMessageType == AWSMessageType.KINESIS_CLOUDWATCH_RAW) {
            final CloudWatchLogSubscriptionData logSubscriptionData = decompressCloudWatchMessages(payloadBytes, objectMapper);
            return logSubscriptionData.logEvents().stream()
                    .map(le -> {
                        DateTime timestamp = new DateTime(le.timestamp(), DateTimeZone.UTC);
                        return KinesisLogEntry.create(kinesisStream,
                                // Use the log group and stream returned from CloudWatch.
                                logSubscriptionData.logGroup(),
                                logSubscriptionData.logStream(), timestamp, le.message());
                    })
                    .collect(Collectors.toList());
        } else if (awsMessageType == AWSMessageType.KINESIS_RAW) {
            // The best timestamp available is the approximate arrival time of the message to the Kinesis stream.
            final DateTime timestamp = new DateTime(approximateArrivalTimestamp.toEpochMilli(), DateTimeZone.UTC);
            final KinesisLogEntry kinesisLogEntry = KinesisLogEntry.create(kinesisStream,
                    "", "",
                    timestamp, new String(payloadBytes, StandardCharsets.UTF_8));
            return Collections.singletonList(kinesisLogEntry);
        } else {
            LOG.error("The AWSMessageType [{}] is not supported by the KinesisTransport", awsMessageType);
            return new ArrayList<>();
        }
    }
@Test
    public void testKinesisRawDecoding() throws IOException {

        final String textLogMessage = "a text log message";
        final Instant now = Instant.now();
        final List<KinesisLogEntry> logEntries =
                rawDecoder.processMessages(textLogMessage.getBytes(StandardCharsets.UTF_8), now);

        Assert.assertEquals(1, logEntries.size());
        // Verify that there are two flow logs present in the parsed result.
        Assert.assertEquals(1, logEntries.stream().filter(logEntry -> logEntry.message().equals(textLogMessage)).count());

        // Verify timestamp and message contents.
        final KinesisLogEntry resultLogEntry = logEntries.stream().findAny().get();
        Assert.assertEquals(textLogMessage, resultLogEntry.message());
        Assert.assertEquals(new DateTime(now.toEpochMilli(), DateTimeZone.UTC), resultLogEntry.timestamp());
    }