KafkaUtils { public static ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset) throws TopicOffsetOutOfRangeException, FailedFetchException,RuntimeException { ByteBufferMessageSet msgs = null; String topic = config.topic; int partitionId = partition.partition; FetchRequestBuilder builder = new FetchRequestBuilder(); FetchRequest fetchRequest = builder.addFetch(topic, partitionId, offset, config.fetchSizeBytes). clientId(config.clientId).maxWait(config.fetchMaxWait).build(); FetchResponse fetchResponse; try { fetchResponse = consumer.fetch(fetchRequest); } catch (Exception e) { if (e instanceof ConnectException || e instanceof SocketTimeoutException || e instanceof IOException || e instanceof UnresolvedAddressException ) { LOG.warn("Network error when fetching messages:", e); throw new FailedFetchException(e); } else { throw new RuntimeException(e); } } if (fetchResponse.hasError()) { KafkaError error = KafkaError.getError(fetchResponse.errorCode(topic, partitionId)); if (error.equals(KafkaError.OFFSET_OUT_OF_RANGE) && config.useStartOffsetTimeIfOffsetOutOfRange) { String msg = "Got fetch request with offset out of range: [" + offset + "]"; LOG.warn(msg); throw new TopicOffsetOutOfRangeException(msg); } else { String message = "Error fetching data from [" + partition + "] for topic [" + topic + "]: [" + error + "]"; LOG.error(message); throw new FailedFetchException(message); } } else { msgs = fetchResponse.messageSet(topic, partitionId); } return msgs; }  static IBrokerReader makeBrokerReader(Map stormConf, KafkaConfig conf); static long getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config); static long getOffset(SimpleConsumer consumer, String topic, int partition, long startOffsetTime); static ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset); static Iterable<List<Object>> generateTuples(KafkaConfig kafkaConfig, Message msg); static List<Partition> calculatePartitionsForTask(GlobalPartitionInformation partitionInformation, int totalTasks, int taskIndex); static String taskId(int taskIndex, int totalTasks); static final Logger LOG; }
@Test(expected = FailedFetchException.class) public void topicDoesNotExist() throws Exception { KafkaUtils.fetchMessages(config, simpleConsumer, new Partition(Broker.fromString(broker.getBrokerConnectionString()), 0), 0); }
@Test(expected = FailedFetchException.class) public void brokerIsDown() throws Exception { int port = broker.getPort(); broker.shutdown(); SimpleConsumer simpleConsumer = new SimpleConsumer("localhost", port, 100, 1024, "testClient"); try { KafkaUtils.fetchMessages(config, simpleConsumer, new Partition(Broker.fromString(broker.getBrokerConnectionString()), 0), OffsetRequest.LatestTime()); } finally { simpleConsumer.close(); } }
@Test(expected = FailedFetchException.class) public void fetchMessagesWithInvalidOffsetAndDefaultHandlingDisabled() throws Exception { config.useStartOffsetTimeIfOffsetOutOfRange = false; KafkaUtils.fetchMessages(config, simpleConsumer, new Partition(Broker.fromString(broker.getBrokerConnectionString()), 0), -99); }
@Test(expected = TopicOffsetOutOfRangeException.class) public void fetchMessagesWithInvalidOffsetAndDefaultHandlingEnabled() throws Exception { config = new KafkaConfig(brokerHosts, "newTopic"); String value = "test"; createTopicAndSendMessage(value); KafkaUtils.fetchMessages(config, simpleConsumer, new Partition(Broker.fromString(broker.getBrokerConnectionString()), 0), -99); }