@Override
    public ExactlyOnceSupport exactlyOnceSupport(Map<String, String> props) {
        return consumerUsesReadCommitted(props)
                ? ExactlyOnceSupport.SUPPORTED
                : ExactlyOnceSupport.UNSUPPORTED;
    }
@Test
    public void testExactlyOnceSupport() {
        String readCommitted = "read_committed";
        String readUncommitted = "read_uncommitted";
        String readGarbage = "read_garbage";

        // Connector is configured correctly, but exactly-once can't be supported
        assertExactlyOnceSupport(null, null, false);
        assertExactlyOnceSupport(readUncommitted, null, false);
        assertExactlyOnceSupport(null, readUncommitted, false);
        assertExactlyOnceSupport(readUncommitted, readUncommitted, false);

        // Connector is configured correctly, and exactly-once can be supported
        assertExactlyOnceSupport(readCommitted, null, true);
        assertExactlyOnceSupport(null, readCommitted, true);
        assertExactlyOnceSupport(readUncommitted, readCommitted, true);
        assertExactlyOnceSupport(readCommitted, readCommitted, true);

        // Connector is configured incorrectly, but is able to react gracefully
        assertExactlyOnceSupport(readGarbage, null, false);
        assertExactlyOnceSupport(null, readGarbage, false);
        assertExactlyOnceSupport(readGarbage, readGarbage, false);
        assertExactlyOnceSupport(readCommitted, readGarbage, false);
        assertExactlyOnceSupport(readUncommitted, readGarbage, false);
        assertExactlyOnceSupport(readGarbage, readUncommitted, false);
        assertExactlyOnceSupport(readGarbage, readCommitted, true);
    }