@Nullable
    @Override
    public Message decode(@Nonnull RawMessage rawMessage) {

        // Load the codec by message type.
        final AWSMessageType awsMessageType = AWSMessageType.valueOf(configuration.getString(CK_AWS_MESSAGE_TYPE));
        final Codec.Factory<? extends Codec> codecFactory = this.availableCodecs.get(awsMessageType.getCodecName());
        if (codecFactory == null) {
            LOG.error("A codec with name [{}] could not be found.", awsMessageType.getCodecName());
            return null;
        }

        final Codec codec = codecFactory.create(configuration);

        // Parse the message with the specified codec.
        final Message message = codec.decode(new RawMessage(rawMessage.getPayload()));
        if (message == null) {
            LOG.error("Failed to decode message for codec [{}].", codec.getName());
            return null;
        }

        return message;
    }
@Test
    public void testKinesisRawCodec() throws JsonProcessingException {

        final HashMap<String, Object> configMap = new HashMap<>();
        configMap.put(AWSCodec.CK_AWS_MESSAGE_TYPE, AWSMessageType.KINESIS_RAW.toString());
        final Configuration configuration = new Configuration(configMap);
        final AWSCodec codec = new AWSCodec(configuration, AWSTestingUtils.buildTestCodecs());

        final DateTime timestamp = DateTime.now(DateTimeZone.UTC);
        final KinesisLogEntry kinesisLogEntry = KinesisLogEntry.create("a-stream", "log-group", "log-stream", timestamp,
                                                                       "This a raw message");

        Message message = codec.decode(new RawMessage(objectMapper.writeValueAsBytes(kinesisLogEntry)));
        Assert.assertEquals("log-group", message.getField(AbstractKinesisCodec.FIELD_LOG_GROUP));
        Assert.assertEquals("log-stream", message.getField(AbstractKinesisCodec.FIELD_LOG_STREAM));
        Assert.assertEquals("a-stream", message.getField(AbstractKinesisCodec.FIELD_KINESIS_STREAM));
        Assert.assertEquals(KinesisRawLogCodec.SOURCE, message.getField("source"));
        Assert.assertEquals("This a raw message", message.getField("message"));
        Assert.assertEquals(timestamp, message.getTimestamp());
    }