@Override
    public void write(IN element, Context context) throws IOException, InterruptedException {
        // setting the values in the bucketer context
        bucketerContext.update(
                context.timestamp(),
                context.currentWatermark(),
                processingTimeService.getCurrentProcessingTime());

        final String bucketId = bucketAssigner.getBucketId(element, bucketerContext);
        final FileWriterBucket<IN> bucket = getOrCreateBucketForBucketId(bucketId);
        bucket.write(element, processingTimeService.getCurrentProcessingTime());
        numRecordsOutCounter.inc();
    }
@Test
    void testNumberRecordsOutCounter(@TempDir java.nio.file.Path tempDir)
            throws IOException, InterruptedException {
        Path path = new Path(tempDir.toUri());

        final OperatorIOMetricGroup operatorIOMetricGroup =
                UnregisteredMetricGroups.createUnregisteredOperatorMetricGroup().getIOMetricGroup();
        final SinkWriterMetricGroup sinkWriterMetricGroup =
                MetricsGroupTestUtils.mockWriterMetricGroup(
                        metricListener.getMetricGroup(), operatorIOMetricGroup);

        Counter recordsCounter = sinkWriterMetricGroup.getIOMetricGroup().getNumRecordsOutCounter();
        SinkWriter.Context context = new ContextImpl();
        FileWriter<String> fileWriter =
                createWriter(
                        path,
                        DefaultRollingPolicy.builder().build(),
                        new OutputFileConfig("part-", ""),
                        sinkWriterMetricGroup);

        assertThat(recordsCounter.getCount()).isEqualTo(0);

        fileWriter.write("1", context);

        assertThat(recordsCounter.getCount()).isEqualTo(1);

        fileWriter.write("2", context);
        fileWriter.write("3", context);

        assertThat(recordsCounter.getCount()).isEqualTo(3);
    }