static Map<String, Object> getOffsetConsumerConfig(
      String name, @Nullable Map<String, Object> offsetConfig, Map<String, Object> consumerConfig) {
    Map<String, Object> offsetConsumerConfig = new HashMap<>(consumerConfig);
    offsetConsumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

    Object groupId = consumerConfig.get(ConsumerConfig.GROUP_ID_CONFIG);
    // override group_id and disable auto_commit so that it does not interfere with main consumer
    String offsetGroupId =
        String.format(
            "%s_offset_consumer_%d_%s",
            name, new Random().nextInt(Integer.MAX_VALUE), (groupId == null ? "none" : groupId));
    offsetConsumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, offsetGroupId);

    if (offsetConfig != null) {
      offsetConsumerConfig.putAll(offsetConfig);
    }

    // Force read isolation level to 'read_uncommitted' for offset consumer. This consumer
    // fetches latest offset for two reasons : (a) to calculate backlog (number of records
    // yet to be consumed) (b) to advance watermark if the backlog is zero. The right thing to do
    // for (a) is to leave this config unchanged from the main config (i.e. if there are records
    // that can't be read because of uncommitted records before them, they shouldn't
    // ideally count towards backlog when "read_committed" is enabled. But (b)
    // requires finding out if there are any records left to be read (committed or uncommitted).
    // Rather than using two separate consumers we will go with better support for (b). If we do
    // hit a case where a lot of records are not readable (due to some stuck transactions), the
    // pipeline would report more backlog, but would not be able to consume it. It might be ok
    // since CPU consumed on the workers would be low and will likely avoid unnecessary upscale.
    offsetConsumerConfig.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_uncommitted");

    return offsetConsumerConfig;
  }
@Test
  public void testOffsetConsumerConfigOverrides() throws Exception {
    KafkaIO.Read<?, ?> spec =
        KafkaIO.read()
            .withBootstrapServers("broker_1:9092,broker_2:9092")
            .withTopic("my_topic")
            .withOffsetConsumerConfigOverrides(null);
    Map<String, Object> offsetConfig =
        KafkaIOUtils.getOffsetConsumerConfig(
            "name", spec.getOffsetConsumerConfig(), spec.getConsumerConfig());
    assertTrue(
        offsetConfig
            .get(ConsumerConfig.GROUP_ID_CONFIG)
            .toString()
            .matches("name_offset_consumer_\\d+_none"));

    assertEquals(false, offsetConfig.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
    assertEquals("read_uncommitted", offsetConfig.get(ConsumerConfig.ISOLATION_LEVEL_CONFIG));

    String offsetGroupId = "group.offsetConsumer";
    KafkaIO.Read<?, ?> spec2 =
        KafkaIO.read()
            .withBootstrapServers("broker_1:9092,broker_2:9092")
            .withTopic("my_topic")
            .withOffsetConsumerConfigOverrides(
                ImmutableMap.of(ConsumerConfig.GROUP_ID_CONFIG, offsetGroupId));
    offsetConfig =
        KafkaIOUtils.getOffsetConsumerConfig(
            "name2", spec2.getOffsetConsumerConfig(), spec2.getConsumerConfig());
    assertEquals(offsetGroupId, offsetConfig.get(ConsumerConfig.GROUP_ID_CONFIG));
    assertEquals(false, offsetConfig.get(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
    assertEquals("read_uncommitted", offsetConfig.get(ConsumerConfig.ISOLATION_LEVEL_CONFIG));
  }