samza-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Chris Riccomini" <criccom...@apache.org>
Subject Re: Review Request 32052: SAMZA-592
Date Fri, 13 Mar 2015 20:47:21 GMT

-----------------------------------------------------------
This is an automatically generated e-mail. To reply, visit:
https://reviews.apache.org/r/32052/#review76431
-----------------------------------------------------------



samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala
<https://reviews.apache.org/r/32052/#comment123974>

    > 1. What would happen to the cache entry if we don't check partitionMetadata.errorCode?
    
    If we don't check partitionMetadata.errorCode *anywhere*, then any partition-level errors
will get triggered when getOffsets is called. For example, if a leader is not available, getOffsets'
OffsetFetchRequest will have an errorCode set that will say the leader isn't available.
    
    We're basically putting off the error check until after we've made the request we care
about (e.g. produce request, offset fetch request, etc).
    
    > 2. Assuming that the errored partition metadata is not inserted in the cache, getOffsets
would raise exception? And how do we capture that case?
    
    Yes, getOffsets calls:
    
          .getOffsetsBefore(new OffsetRequest(partitionOffsetInfo))
    
    And then checks ths results:
    
            KafkaUtil.maybeThrowException(partitionErrorAndOffset.error)
    
    Any bad errorCodes from the metadata will also get surfaced here (e.g. unknown topic/partition,
leader offline, etc).
    
    If there is a failure, the maybeThrowException method will throw an exception, which will
be caught in KafkaSystemAdmin.getSystemStreamMetadata. The retryBackoff loop will then catch
the exception, and restart from scratch.


- Chris Riccomini


On March 13, 2015, 7:56 p.m., Chris Riccomini wrote:
> 
> -----------------------------------------------------------
> This is an automatically generated e-mail. To reply, visit:
> https://reviews.apache.org/r/32052/
> -----------------------------------------------------------
> 
> (Updated March 13, 2015, 7:56 p.m.)
> 
> 
> Review request for samza.
> 
> 
> Bugs: SAMZA-592
>     https://issues.apache.org/jira/browse/SAMZA-592
> 
> 
> Repository: samza
> 
> 
> Description
> -------
> 
> refresh topic metadata if partitions have bad error codes. add a test
> 
> 
> add a little test to verify we ignore replica not available exceptions
> 
> 
> remove partition metadata check from KafkaSystemAdmin since it's already done in getOffsets
> 
> 
> switch to KafkaUtil.maybeThrowException
> 
> 
> Diffs
> -----
> 
>   samza-kafka/src/main/scala/org/apache/samza/checkpoint/kafka/KafkaCheckpointManager.scala
4a1b31f025ba7b05a7b46041aa8e12074599ce24 
>   samza-kafka/src/main/scala/org/apache/samza/system/kafka/BrokerProxy.scala c6e231a2588ce95940aa2da9483a98c6115e38d9

>   samza-kafka/src/main/scala/org/apache/samza/system/kafka/GetOffset.scala 147aabc947f0cb01c0780edb693e9714f810b5f6

>   samza-kafka/src/main/scala/org/apache/samza/system/kafka/KafkaSystemAdmin.scala b790be17cfe08da28220ffb381cbd618ebe25cf0

>   samza-kafka/src/main/scala/org/apache/samza/system/kafka/TopicMetadataCache.scala 4a49d22a3fc403f624ca17a6414d84eaba1898be

>   samza-kafka/src/main/scala/org/apache/samza/util/KafkaUtil.scala 2482f23cc6b9c072651df9cbfe9714ffeb203687

>   samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala
3d1e6ecbb3fd95816c722a68c4f5907120eb20d0 
>   samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestTopicMetadataCache.scala
e698d2f1f004740a4d74a488c469d8ca8426c6e4 
>   samza-kafka/src/test/scala/org/apache/samza/utils/TestKafkaUtil.scala PRE-CREATION

>   samza-test/src/test/scala/org/apache/samza/test/integration/TestStatefulTask.scala
a8b724bf781003142e455fdf1fed2f13d6c18353 
> 
> Diff: https://reviews.apache.org/r/32052/diff/
> 
> 
> Testing
> -------
> 
> 
> Thanks,
> 
> Chris Riccomini
> 
>


Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message