KafkaConsumerConfig extends HashMap<String, Object> { static String getAutoOffsetResetValue(final String autoOffsetReset, final String samzaOffsetDefault) { final String kafkaOffsetLatest = "latest"; final String kafkaOffsetEarliest = "earliest"; final String kafkaOffsetNone = "none"; if (!StringUtils.isBlank(autoOffsetReset)) { if (autoOffsetReset.equals(kafkaOffsetEarliest) || autoOffsetReset.equals(kafkaOffsetLatest) || autoOffsetReset.equals(kafkaOffsetNone)) { return autoOffsetReset; } String newAutoOffsetReset; switch (autoOffsetReset) { case "largest": newAutoOffsetReset = kafkaOffsetLatest; LOG.warn("Using old (deprecated) value for kafka consumer config auto.offset.reset = {}. The right value should be {}", autoOffsetReset, kafkaOffsetLatest); break; case "smallest": newAutoOffsetReset = kafkaOffsetEarliest; LOG.warn("Using old (deprecated) value for kafka consumer config auto.offset.reset = {}. The right value should be {}", autoOffsetReset, kafkaOffsetEarliest); break; default: throw new SamzaException("Using invalid value for kafka consumer config auto.offset.reset " + autoOffsetReset + ". See KafkaConsumer config for the correct values."); } LOG.info("Auto offset reset value converted from {} to {}", autoOffsetReset, newAutoOffsetReset); return newAutoOffsetReset; } String newAutoOffsetReset = kafkaOffsetLatest; if (!StringUtils.isBlank(samzaOffsetDefault)) { switch (samzaOffsetDefault) { case SystemConfig.SAMZA_SYSTEM_OFFSET_UPCOMING: newAutoOffsetReset = kafkaOffsetLatest; break; case SystemConfig.SAMZA_SYSTEM_OFFSET_OLDEST: newAutoOffsetReset = kafkaOffsetEarliest; break; default: throw new SamzaException("Using invalid value for samza default offset config " + autoOffsetReset + ". See samza config for the correct values"); } LOG.info("Auto offset reset value for KafkaConsumer for system {} converted from {}(samza) to {}", samzaOffsetDefault, newAutoOffsetReset); } return newAutoOffsetReset; } private  KafkaConsumerConfig(Map<String, Object> props, String systemName); static KafkaConsumerConfig getKafkaSystemConsumerConfig(Config config, String systemName, String clientId); String getClientId(); int fetchMessageMaxBytes(); String getZkConnect(); static String createClientId(String prefix, Config config); static Pair<String, String> getJobNameAndId(Config config); static final Logger LOG; static final String ZOOKEEPER_CONNECT; }
@Test public void testKafkaAutoResetValue() { Assert.assertEquals("latest", KafkaConsumerConfig.getAutoOffsetResetValue("latest", "oldest")); try { KafkaConsumerConfig.getAutoOffsetResetValue("someValue", "oldest"); Assert.fail("Invalid value should've triggered an exception"); } catch (Exception e) { } Assert.assertEquals("earliest", KafkaConsumerConfig.getAutoOffsetResetValue("earliest", "upcoming")); Assert.assertEquals("none", KafkaConsumerConfig.getAutoOffsetResetValue("none", "oldest")); Assert.assertEquals("latest", KafkaConsumerConfig.getAutoOffsetResetValue("largest", "oldest")); Assert.assertEquals("earliest", KafkaConsumerConfig.getAutoOffsetResetValue("smallest", "upcoming")); Assert.assertEquals("earliest", KafkaConsumerConfig.getAutoOffsetResetValue("", "oldest")); Assert.assertEquals("latest", KafkaConsumerConfig.getAutoOffsetResetValue("", "upcoming")); try { KafkaConsumerConfig.getAutoOffsetResetValue("", "whatever"); Assert.fail("Invalid value should've triggered an exception"); } catch (Exception e) { } }