KafkaSource extends AbstractPollableSource implements Configurable, BatchSizeSupported { @Override protected void doConfigure(Context context) throws FlumeException { this.context = context; headers = new HashMap<String, String>(4); tpAndOffsetMetadata = new HashMap<TopicPartition, OffsetAndMetadata>(); rebalanceFlag = new AtomicBoolean(false); kafkaProps = new Properties(); translateOldProperties(context); String topicProperty = context.getString(KafkaSourceConstants.TOPICS_REGEX); if (topicProperty != null && !topicProperty.isEmpty()) { subscriber = new PatternSubscriber(topicProperty); } else if ((topicProperty = context.getString(KafkaSourceConstants.TOPICS)) != null && !topicProperty.isEmpty()) { subscriber = new TopicListSubscriber(topicProperty); } else if (subscriber == null) { throw new ConfigurationException("At least one Kafka topic must be specified."); } batchUpperLimit = context.getInteger(KafkaSourceConstants.BATCH_SIZE, KafkaSourceConstants.DEFAULT_BATCH_SIZE); maxBatchDurationMillis = context.getInteger(KafkaSourceConstants.BATCH_DURATION_MS, KafkaSourceConstants.DEFAULT_BATCH_DURATION); useAvroEventFormat = context.getBoolean(KafkaSourceConstants.AVRO_EVENT, KafkaSourceConstants.DEFAULT_AVRO_EVENT); if (log.isDebugEnabled()) { log.debug(KafkaSourceConstants.AVRO_EVENT + " set to: {}", useAvroEventFormat); } zookeeperConnect = context.getString(ZOOKEEPER_CONNECT_FLUME_KEY); migrateZookeeperOffsets = context.getBoolean(MIGRATE_ZOOKEEPER_OFFSETS, DEFAULT_MIGRATE_ZOOKEEPER_OFFSETS); bootstrapServers = context.getString(KafkaSourceConstants.BOOTSTRAP_SERVERS); if (bootstrapServers == null || bootstrapServers.isEmpty()) { if (zookeeperConnect == null || zookeeperConnect.isEmpty()) { throw new ConfigurationException("Bootstrap Servers must be specified"); } else { log.warn("{} is deprecated. Please use the parameter {}", KafkaSourceConstants.ZOOKEEPER_CONNECT_FLUME_KEY, KafkaSourceConstants.BOOTSTRAP_SERVERS); String securityProtocolStr = context.getSubProperties(KafkaSourceConstants.KAFKA_CONSUMER_PREFIX) .get(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG); if (securityProtocolStr == null || securityProtocolStr.isEmpty()) { securityProtocolStr = CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL; } bootstrapServers = lookupBootstrap(zookeeperConnect, SecurityProtocol.valueOf(securityProtocolStr)); } } String groupIdProperty = context.getString(KAFKA_CONSUMER_PREFIX + ConsumerConfig.GROUP_ID_CONFIG); if (groupIdProperty != null && !groupIdProperty.isEmpty()) { groupId = groupIdProperty; } if (groupId == null || groupId.isEmpty()) { groupId = DEFAULT_GROUP_ID; log.info("Group ID was not specified. Using {} as the group id.", groupId); } setTopicHeader = context.getBoolean(KafkaSourceConstants.SET_TOPIC_HEADER, KafkaSourceConstants.DEFAULT_SET_TOPIC_HEADER); topicHeader = context.getString(KafkaSourceConstants.TOPIC_HEADER, KafkaSourceConstants.DEFAULT_TOPIC_HEADER); setConsumerProps(context); if (log.isDebugEnabled() && LogPrivacyUtil.allowLogPrintConfig()) { log.debug("Kafka consumer properties: {}", kafkaProps); } if (counter == null) { counter = new KafkaSourceCounter(getName()); } }  @Override long getBatchSize();  }
@Test public void testMigrateZookeeperOffsetsWhenTopicNotExists() throws Exception { String topic = findUnusedTopic(); Context context = prepareDefaultContext("testMigrateOffsets-nonExistingTopic"); context.put(ZOOKEEPER_CONNECT_FLUME_KEY, kafkaServer.getZkConnectString()); context.put(TOPIC, topic); KafkaSource source = new KafkaSource(); source.doConfigure(context); source.setChannelProcessor(createGoodChannel()); source.start(); assertEquals(LifecycleState.START, source.getLifecycleState()); Status status = source.process(); assertEquals(Status.BACKOFF, status); source.stop(); }