@Override
    public PipelineJobProgressUpdatedParameter write(final String ackId, final Collection<Record> records) {
        if (records.isEmpty()) {
            return new PipelineJobProgressUpdatedParameter(0);
        }
        while (!channel.isWritable() && channel.isActive()) {
            doAwait();
        }
        if (!channel.isActive()) {
            return new PipelineJobProgressUpdatedParameter(0);
        }
        Collection<DataRecordResult.Record> resultRecords = getResultRecords(records);
        DataRecordResult dataRecordResult = DataRecordResult.newBuilder().addAllRecord(resultRecords).setAckId(ackId).build();
        channel.writeAndFlush(CDCResponseUtils.succeed("", ResponseCase.DATA_RECORD_RESULT, dataRecordResult));
        return new PipelineJobProgressUpdatedParameter(resultRecords.size());
    }
@Test
    void assertWrite() throws IOException {
        Channel mockChannel = mock(Channel.class);
        when(mockChannel.isWritable()).thenReturn(false, true);
        when(mockChannel.isActive()).thenReturn(true);
        ShardingSphereDatabase mockDatabase = mock(ShardingSphereDatabase.class);
        when(mockDatabase.getName()).thenReturn("test");
        try (PipelineCDCSocketSink sink = new PipelineCDCSocketSink(mockChannel, mockDatabase, Collections.singletonList("test.t_order"))) {
            PipelineJobProgressUpdatedParameter actual = sink.write("ack", Collections.singletonList(new FinishedRecord(new IngestPlaceholderPosition())));
            assertThat(actual.getProcessedRecordsCount(), is(0));
            actual = sink.write("ack", Collections.singletonList(new DataRecord(PipelineSQLOperationType.DELETE, "t_order", new IngestPlaceholderPosition(), 1)));
            assertThat(actual.getProcessedRecordsCount(), is(1));
        }
    }