KafkaIOUtils { static Map<String, Object> getOffsetConsumerConfig( String name, 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); 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); } 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)); }