@Override
    public void write(Record<GenericObject> record) throws Exception {
        // kpl-thread captures publish-failure. fail the publish on main pulsar-io-thread to maintain the ordering
        if (kinesisSinkConfig.isRetainOrdering() && previousPublishFailed == TRUE) {
            LOG.warn("Skip acking message to retain ordering with previous failed message {}-{}", this.streamName,
                    record.getRecordSequence());
            throw new IllegalStateException("kinesis queue has publish failure");
        }
        String partitionedKey = record.getKey().orElse(record.getTopicName().orElse(defaultPartitionedKey));
        partitionedKey = partitionedKey.length() > maxPartitionedKeyLength
                ? partitionedKey.substring(0, maxPartitionedKeyLength - 1)
                : partitionedKey; // partitionedKey Length must be at least one, and at most 256
        ByteBuffer data = createKinesisMessage(kinesisSinkConfig.getMessageFormat(), record);
        int size = data.remaining();
        sendUserRecord(ProducerSendCallback.create(this, record, System.nanoTime(), partitionedKey, data));
        if (sinkContext != null) {
            sinkContext.recordMetric(METRICS_TOTAL_INCOMING, 1);
            sinkContext.recordMetric(METRICS_TOTAL_INCOMING_BYTES, data.array().length);
        }
        if (LOG.isDebugEnabled()) {
            LOG.debug("Published message to kinesis stream {} with size {}", streamName, size);
        }
    }
@Test
    public void testWrite() throws Exception {
        AtomicBoolean ackCalled = new AtomicBoolean();
        AtomicLong sequenceCounter = new AtomicLong(0);
        Message<GenericObject> message = mock(Message.class);
        when(message.getData()).thenReturn("hello".getBytes(StandardCharsets.UTF_8));
        final Record<GenericObject> pulsarRecord = new Record<GenericObject>() {

            @Override
            public Optional<String> getKey() {
                return Optional.of( "key-" + sequenceCounter.incrementAndGet());
            }

            @Override
            public GenericObject getValue() {
                // Value comes from the message raw data, not the GenericObject
                return null;
            }

            @Override
            public void ack() {
                ackCalled.set(true);
            }

            @Override
            public Optional<Message<GenericObject>> getMessage() {
                return Optional.of(message);
            }
        };

        try (final KinesisSink sink = new KinesisSink()) {
            Map<String, Object> map = createConfig();
            SinkContext mockSinkContext = mock(SinkContext.class);

            sink.open(map, mockSinkContext);
            for (int i = 0; i < 10; i++) {
                sink.write(pulsarRecord);
            }
            Awaitility.await().untilAsserted(() -> {
                assertTrue(ackCalled.get());
            });
            final GetRecordsResponse getRecords = getStreamRecords();
            assertEquals(getRecords.records().size(), 10);

            for (software.amazon.awssdk.services.kinesis.model.Record record : getRecords.records()) {
                assertEquals(record.data().asString(StandardCharsets.UTF_8), "hello");
            }
        }
    }