Build failed in Jenkins: kafka-trunk-jdk8 #3871

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-8412: Fix nullpointer exception thrown on flushing before 
closing

[harsha] KAFKA-8669: Add security providers in kafka security config (#7090)

--
[...truncated 8.51 MB...]

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testAlreadyFlushing 
PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testCancelBeforeAwaitFlush STARTED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testCancelBeforeAwaitFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testCancelAfterAwaitFlush STARTED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testCancelAfterAwaitFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testWriteFlush 
STARTED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testWriteFlush PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testPutConnectorConfig STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testPutConnectorConfig PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testPutTaskConfigs STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testPutTaskConfigs PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testPutTaskConfigsZeroTasks STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testPutTaskConfigsZeroTasks PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testRestoreTargetState STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testRestoreTargetState PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testBackgroundUpdateTargetState STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testBackgroundUpdateTargetState PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testBackgroundConnectorDeletion STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testBackgroundConnectorDeletion PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > testRestore 
STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > testRestore 
PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testRestoreConnectorDeletion STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testRestoreConnectorDeletion PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testRestoreZeroTasks STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testRestoreZeroTasks PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testPutTaskConfigsDoesNotResolveAllInconsistencies STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testPutTaskConfigsDoesNotResolveAllInconsistencies PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > testStartStop 
STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > testStartStop 
PASSED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testRestoreTargetStateUnexpectedDeletion STARTED

org.apache.kafka.connect.storage.KafkaConfigBackingStoreTest > 
testRestoreTargetStateUnexpectedDeletion PASSED

org.apache.kafka.connect.storage.FileOffsetBackingStoreTest > testSaveRestore 
STARTED

org.apache.kafka.connect.storage.FileOffsetBackingStoreTest > testSaveRestore 
PASSED

org.apache.kafka.connect.storage.FileOffsetBackingStoreTest > testGetSet STARTED

org.apache.kafka.connect.storage.FileOffsetBackingStoreTest > testGetSet PASSED

org.apache.kafka.connect.integration.ErrorHandlingIntegrationTest > 
testSkipRetryAndDLQWithHeaders STARTED

org.apache.kafka.connect.integration.ErrorHandlingIntegrationTest > 
testSkipRetryAndDLQWithHeaders PASSED

org.apache.kafka.connect.integration.StartAndStopLatchTest > 
shouldReturnFalseWhenAwaitingForDependentLatchToComplete STARTED

org.apache.kafka.connect.integration.StartAndStopLatchTest > 
shouldReturnFalseWhenAwaitingForDependentLatchToComplete PASSED

org.apache.kafka.connect.integration.StartAndStopLatchTest > 
shouldReturnTrueWhenAwaitingForStartAndStopAndDependentLatch STARTED

org.apache.kafka.connect.integration.StartAndStopLatchTest > 
shouldReturnTrueWhenAwaitingForStartAndStopAndDependentLatch PASSED

org.apache.kafka.connect.integration.StartAndStopLatchTest > 
shouldReturnFalseWhenAwaitingForStartToNeverComplete STARTED

org.apache.kafka.connect.integration.StartAndStopLatchTest > 
shouldReturnFalseWhenAwaitingForStartToNeverComplete PASSED

org.apache.kafka.connect.integration.StartAndStopLatchTest > 
shouldReturnFalseWhenAwaitingForStopToNeverComplete STARTED

org.apache.kafka.connect.integration.StartAndStopLatchTest > 
shouldReturnFalseWhenAwaitingForStopToNeverComplete PASSED

org.apache.kafka.connect.integration.StartAndStopLatchTest > 

Build failed in Jenkins: kafka-trunk-jdk11 #776

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-8816: Make offsets immutable to users of 
RecordCollector.offsets

--
[...truncated 2.61 MB...]
org.apache.kafka.streams.integration.SuppressionIntegrationTest > 
shouldShutdownWhenBytesConstraintIsViolated STARTED

org.apache.kafka.streams.integration.SuppressionIntegrationTest > 
shouldShutdownWhenBytesConstraintIsViolated PASSED

org.apache.kafka.streams.integration.GlobalKTableEOSIntegrationTest > 
shouldKStreamGlobalKTableLeftJoin STARTED

org.apache.kafka.streams.integration.GlobalKTableEOSIntegrationTest > 
shouldKStreamGlobalKTableLeftJoin PASSED

org.apache.kafka.streams.integration.GlobalKTableEOSIntegrationTest > 
shouldNotRestoreAbortedMessages STARTED

org.apache.kafka.streams.integration.GlobalKTableEOSIntegrationTest > 
shouldNotRestoreAbortedMessages PASSED

org.apache.kafka.streams.integration.GlobalKTableEOSIntegrationTest > 
shouldRestoreTransactionalMessages STARTED

org.apache.kafka.streams.integration.GlobalKTableEOSIntegrationTest > 
shouldRestoreTransactionalMessages PASSED

org.apache.kafka.streams.integration.GlobalKTableEOSIntegrationTest > 
shouldKStreamGlobalKTableJoin STARTED

org.apache.kafka.streams.integration.GlobalKTableEOSIntegrationTest > 
shouldKStreamGlobalKTableJoin PASSED

org.apache.kafka.streams.integration.StandbyTaskCreationIntegrationTest > 
shouldNotCreateAnyStandByTasksForStateStoreWithLoggingDisabled STARTED

org.apache.kafka.streams.integration.StandbyTaskCreationIntegrationTest > 
shouldNotCreateAnyStandByTasksForStateStoreWithLoggingDisabled PASSED

org.apache.kafka.streams.integration.StandbyTaskCreationIntegrationTest > 
shouldCreateStandByTasksForMaterializedAndOptimizedSourceTables STARTED

org.apache.kafka.streams.integration.StandbyTaskCreationIntegrationTest > 
shouldCreateStandByTasksForMaterializedAndOptimizedSourceTables PASSED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldOnlyReadRecordsWhereEarliestSpecifiedWithNoCommittedOffsetsWithGlobalAutoOffsetResetLatest
 STARTED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldOnlyReadRecordsWhereEarliestSpecifiedWithNoCommittedOffsetsWithGlobalAutoOffsetResetLatest
 PASSED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldThrowExceptionOverlappingPattern STARTED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldThrowExceptionOverlappingPattern PASSED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldThrowExceptionOverlappingTopic STARTED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldThrowExceptionOverlappingTopic PASSED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldOnlyReadRecordsWhereEarliestSpecifiedWithInvalidCommittedOffsets STARTED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldOnlyReadRecordsWhereEarliestSpecifiedWithInvalidCommittedOffsets PASSED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldOnlyReadRecordsWhereEarliestSpecifiedWithNoCommittedOffsetsWithDefaultGlobalAutoOffsetResetEarliest
 STARTED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldOnlyReadRecordsWhereEarliestSpecifiedWithNoCommittedOffsetsWithDefaultGlobalAutoOffsetResetEarliest
 PASSED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldThrowStreamsExceptionNoResetSpecified STARTED

org.apache.kafka.streams.integration.FineGrainedAutoResetIntegrationTest > 
shouldThrowStreamsExceptionNoResetSpecified PASSED

org.apache.kafka.streams.integration.GlobalThreadShutDownOrderTest > 
shouldFinishGlobalStoreOperationOnShutDown STARTED

org.apache.kafka.streams.integration.GlobalThreadShutDownOrderTest > 
shouldFinishGlobalStoreOperationOnShutDown PASSED

org.apache.kafka.streams.integration.ResetIntegrationTest > 
shouldNotAllowToResetWhenInputTopicAbsent STARTED

org.apache.kafka.streams.integration.ResetIntegrationTest > 
shouldNotAllowToResetWhenInputTopicAbsent PASSED

org.apache.kafka.streams.integration.ResetIntegrationTest > 
shouldNotAllowToResetWhenIntermediateTopicAbsent STARTED

org.apache.kafka.streams.integration.ResetIntegrationTest > 
shouldNotAllowToResetWhenIntermediateTopicAbsent PASSED

org.apache.kafka.streams.integration.ResetIntegrationTest > 
testReprocessingByDurationAfterResetWithoutIntermediateUserTopic STARTED

org.apache.kafka.streams.integration.ResetIntegrationTest > 
testReprocessingByDurationAfterResetWithoutIntermediateUserTopic PASSED

org.apache.kafka.streams.integration.ResetIntegrationTest > 
testReprocessingFromScratchAfterResetWithIntermediateUserTopic STARTED

org.apache.kafka.streams.integration.ResetIntegrationTest > 

Re: [DISCUSS] KIP-360: Improve handling of unknown producer

[Guozhang Wang]

Hi Jason,

I've made another pass on the wiki page and it reads much better now. One
more clarification about the "Simplified error handling" section:

1. There will be no "retriable error" from the broker side regarding any
send requests and txn requests (to txn coordinators). All errors would
cause the corresponding txn to eventually be aborted.
2. Some errors (UNKNOWN_PRODUCER, INVALID_PID_MAPPING) would cause the
producer entering the ABORTABLE_ERROR state, but only the current txn to be
aborted; some others (INVALID_PRODUCER_EPOCH) would cause the producer to
enter the FATAL_ERROR state, plus it would cause all future txns to be
aborted.

Is that right?


Guozhang


On Wed, Aug 21, 2019 at 3:52 PM Matthias J. Sax 
wrote:
>
> Thanks Jason!
>
> LGTM.
>
> On 8/21/19 3:07 PM, Jason Gustafson wrote:
> > Hi Matthias,
> >
> > Thanks, I appreciate the thorough review. I've revised the section to
make
> > the logic clearer. I think you have it right except for the 1). We only
> > generate a new PID if the epoch cannot be incremented without overflow.
> >
> > -Jason
> >
> > On Tue, Aug 20, 2019 at 5:45 PM Matthias J. Sax 
> > wrote:
> >
> >> Thanks for the KIP. I just have some clarification questions to make
> >> sure I understand the proposal correctly:
> >>
> >> 1) "Safe Epoch Incrementing"
> >>
> >>> When the coordinator receives a new InitProducerId request, we will
use
> >> the following logic to update the epoch:
> >>>
> >>> 1. No epoch is provided: the current epoch will be bumped and the last
> >> epoch will be set to -1.
> >>> 2. Epoch and producerId are provided, and the provided producerId
> >> matches the current producerId or the provided producerId matches the
> >> previous producerId and the provided epoch is exhausted:
> >>>   a. Provided epoch matches current epoch: the last epoch will be
> >> set to the current epoch, and the current epoch will be bumped .
> >>>   b. Provided epoch matches last epoch: the current epoch will be
> >> returned
> >>>   c. Else: return INVALID_PRODUCER_EPOCH
> >>> 3. Otherwise, return INVALID_PRODUCER_EPOCH
> >>
> >> Case (1) would be for a new producer. Hence, should we state that "no
> >> PID" is provided (instead of "no epoch" is provided?). That might be
> >> clearer and it implies that there is no epoch anyway.
> >>
> >> Case (2) would be for a producer that received `UNKNOWN_PRODUCER_ID`
> >> error and tries to re-initialize itself.
> >>
> >> Case (2a) implies that the producer send its first request and is not
> >> fenced. Case (2b) implies that the producer re-tries to re-initialize
> >> itself, ie, it first request to re-initilize did not get a respond but
> >> was processed by the transaction coordinator. Case (2c) implies that a
> >> producer was fenced (similar case 3, even if I am not sure what case 3
> >> actually would be?)
> >>
> >> Please let me know if my understanding is correct.
> >>
> >> What is still unclear to me is, why case (2 -- or is it only 2b?)
> >> requires that the "provide epoch is exhausted"?
> >>
> >> For case 2b:
> >>
> >> Assume a producer with PID=100 and epoch=MAX_EPOCH that gets an
> >> `UNKNOWN_PRODUCER_ID`. It sends initPidRequest with the corresponding
> >> PID/epoch pair. The TC processes the request and creates a new PID=101
> >> with new epoch=0, however, the respond to the producer is lost. The TC
> >> still stores `currentPid=101`, `currentEpoch=0` and `previousPid=100`,
> >> `previoudEpoch=MAX_EPOCH`. When the producer re-tries, the sent
> >> PID/epoch still matches the previous PID/epoch pair and hence the TC
> >> know it's a retry?
> >>
> >> If this reasoning is correct, should the logic be as follows:
> >>
> >> 1. No PID is provided: create a new PID with epoch=0 and set the last
> >> epoch to -1.
> >> 2. Epoch and producerId are provided
> >>a) the provided producerId/epoch matches the current
producerId/epoch:
> >>   i) if the epoch is not exhausted, bump the epoch
> >>   ii) if the epoch is exhausted, create a new PID with epoch=0
> >>b) the provided producerId/epoch matches the previous
> >> producerId/epoch: respond with current PID/epoch
> >>c) Otherwise, return INVALID_PRODUCER_EPOCH
> >>
> >>
> >>
> >> -Matthias
> >>
> >>
> >>
> >>
> >> On 4/4/19 3:47 PM, Jason Gustafson wrote:
> >>> Hi Everyone,
> >>>
> >>> Sorry for the long delay on this KIP. I have updated it to include the
> >>> handling of INVALID_PRODUCER_ID_MAPPING as suggested above. If there
are
> >> no
> >>> further comments, I will plan to start a vote early next week.
> >>>
> >>> Thanks!
> >>> Jason
> >>>
> >>> On Mon, Mar 25, 2019 at 2:33 PM Adam Bellemare <
adam.bellem...@gmail.com
> >>>
> >>> wrote:
> >>>
>  Ach - Sorry. I meant Jason. I had just read a John Roesler email.
> 
>  On Mon, Mar 25, 2019 at 5:21 PM Adam Bellemare <
> >> adam.bellem...@gmail.com>
>  wrote:
> 
> > Hi John
> >
> > What is the status of this KIP?
> >
> > My teammates and I are running into the 

Re: [DISCUSS] KIP-500: Replace ZooKeeper with a Self-Managed Metadata Quorum

[Colin McCabe]

Hi Ryanne,

Good point.  I added a section titled "future work" with information about the 
follow-on KIPs that we discussed here.

best,
Colin


On Fri, Aug 23, 2019, at 13:15, Ryanne Dolan wrote:
> Thanks Colin, sgtm. Please make this clear in the KIP -- otherwise it is
> hard to nail down what we are voting for.
> 
> Ryanne
> 
> 
> On Fri, Aug 23, 2019, 12:58 PM Colin McCabe  wrote:
> 
> > On Fri, Aug 23, 2019, at 06:24, Ryanne Dolan wrote:
> > > Colin, can you outline what specifically would be in scope for this KIP
> > vs
> > > deferred to the follow-on KIPs you've mentioned? Maybe a Future Work
> > > section? Is the idea to get to the bridge release with this KIP, and then
> > > go from there?
> > >
> > > Ryanne
> > >
> >
> > Hi Ryanne,
> >
> > The goal for KIP-500 is to set out an overall vision for how we will
> > remove ZooKeeper and transition to managing metadata via a controller
> > quorum.
> >
> > We will create follow-on KIPs that will lay out the specific details of
> > each step.
> >
> > * A KIP for allowing kafka-configs.sh to change topic configurations
> > without using ZooKeeper.  (It can already change broker configurations
> > without ZK)
> >
> > * A KIP for adding APIs to replace direct ZK access by the brokers.
> >
> > * A KIP to describe Raft replication in Kafka, including the overall
> > protocol, details of each RPC, etc.
> >
> > * A KIP describing the controller changes, how metadata is stored, etc.
> >
> > There may be other KIPs that we need (for example, if we find another tool
> > that still has a hard ZK dependency), but that's the general idea.  KIP-500
> > is about the overall design-- the follow on KIPs are about the specific
> > details.
> >
> > best,
> > Colin
> >
> >
> > >
> > > On Thu, Aug 22, 2019, 11:58 AM Colin McCabe  wrote:
> > >
> > > > On Wed, Aug 21, 2019, at 19:48, Ron Dagostino wrote:
> > > > > Thanks, Colin.  The changes you made to the KIP related to the bridge
> > > > > release help make it clearer.  I still have some confusion about the
> > > > phrase
> > > > > "The rolling upgrade from the bridge release will take several
> > steps."
> > > > > This made me think you are talking about moving from the bridge
> > release
> > > > to
> > > > > some other, newer, release that comes after the bridge release.  But
> > I
> > > > > think what you are getting at is that the bridge release can be run
> > with
> > > > or
> > > > > without Zookeeper -- when first upgrading to it Zookeeper remains in
> > use,
> > > > > but then there is a transition that can be made to engage the warp
> > > > drive...
> > > > > I mean the Controller Quorum.  So maybe the phrase should be "The
> > rolling
> > > > > upgrade through the bridge release -- starting with Zookeeper being
> > in
> > > > use
> > > > > and ending with Zookeeper having been replaced by the Controller
> > Quorum
> > > > --
> > > > > will take several steps."
> > > >
> > > > Hi Ron,
> > > >
> > > > To clarify, the bridge release will require ZooKeeper.  It will also
> > not
> > > > support the controller quorum.  It's a bridge in the sense that you
> > must
> > > > upgrade to a bridge release prior to upgrading to a ZK-less release.  I
> > > > added some more descriptive text to the bridge release paragraph--
> > > > hopefully this makes it clearer.
> > > >
> > > > best,
> > > > Colin
> > > >
> > > > >
> > > > > Do I understand it correctly, and might some change in phrasing or
> > > > > additional clarification help others avoid the same confusion I had?
> > > > >
> > > > > Ron
> > > > >
> > > > > On Wed, Aug 21, 2019 at 2:31 PM Colin McCabe 
> > wrote:
> > > > >
> > > > > > On Wed, Aug 21, 2019, at 04:22, Ron Dagostino wrote:
> > > > > > > Hi Colin.  I like the concept of a "bridge release" for migrating
> > > > off of
> > > > > > > Zookeeper, but I worry that it may become a bottleneck if people
> > > > hesitate
> > > > > > > to replace Zookeeper -- they would be unable to adopt newer
> > versions
> > > > of
> > > > > > > Kafka until taking (what feels to them like) a giant leap.  As an
> > > > > > example,
> > > > > > > assuming version 4.0.x of Kafka is the supported bridge release,
> > I
> > > > would
> > > > > > > not be surprised if uptake of the 4.x release and the time-based
> > > > releases
> > > > > > > that follow it end up being much slower due to the perceived
> > barrier.
> > > > > > >
> > > > > > > Any perceived barrier could be lowered if the 4.0.x release could
> > > > > > > optionally continue to use Zookeeper -- then the cutover would
> > be two
> > > > > > > incremental steps (move to 4.0.x, then replace Zookeeper while
> > > > staying on
> > > > > > > 4.0.x) as opposed to a single big-bang (upgrade to 4.0.x and
> > replace
> > > > > > > Zookeeper in one fell swoop).
> > > > > >
> > > > > > Hi Ron,
> > > > > >
> > > > > > Just to clarify, the "bridge release" will continue to use
> > ZooKeeper.
> > > > It
> > > > > > will not support running without ZooKeeper.  It is the releases
> > that

[jira] [Resolved] (KAFKA-8831) Joining a new instance sometimes does not cause rebalancing

[Vinoth Chandar (Jira)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-8831?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Vinoth Chandar resolved KAFKA-8831.
---
Resolution: Not A Problem

I will think about better logging and put up a patch. Closing this issue

> Joining a new instance sometimes does not cause rebalancing
> ---
>
> Key: KAFKA-8831
> URL: https://issues.apache.org/jira/browse/KAFKA-8831
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Reporter: Chris Pettitt
>Assignee: Chris Pettitt
>Priority: Major
> Attachments: StandbyTaskTest.java, fail.log
>
>
> See attached log. The application is in a REBALANCING state. The second 
> instance joins a bit after the first instance (~250ms). The group coordinator 
> says it is going to rebalance but nothing happens. The first instance gets 
> all partitions (2). The application transitions to RUNNING.
> See attached test, which starts one client and then starts another about 
> 250ms later. This seems to consistently repro the issue for me.
> This is blocking my work on KAFKA-8755, so I'm inclined to pick it up



--
This message was sent by Atlassian Jira
(v8.3.2#803003)

Re: [ DISCUSS ] KIP-512:Adding headers to RecordMetaData

[Renuka M]

Hi Eric,

We thought about that but we didn't find the strong  enough reason for
having record itself in Acknowledgement.
Headers are supposed to carry metadata and that is the reason headers are
added to producer/consumer records.
Also we feel having headers information in record metadata is good enough
to bridge the gap and link the record to its metadata.
Its simple change since we are not adding any new method signatures.
Adding new method signatures requires adoption and deprecation of old ones
to reduce duplication.
If we get enough votes on adding new method signature, we are open to add
it.

Thanks
Renuka M

On Mon, Aug 26, 2019 at 10:54 AM Eric Azama  wrote:

> Have you considered adding a new onAcknowledgement method to the
> ProducerInterceptor with the signature onAcknowledgement(RecordMetadata
> metadata, Exception exception, ProducerRecord record)? I would also
> consider adding this to Producer Callbacks as well, since linking a
> Callback to a specific record currently requires creating a new Callback
> for every ProducerRecord sent.
>
> This seems like a more robust strategy compared to using Headers. Headers
> don't necessarily contain anything that connects them to the original
> ProducerRecord, and forcibly including information in the Headers seems
> like unnecessary bloat. If your goal is to link a RecordMetadata to a
> specific ProducerRecord, it seems simpler to make sure the original
> ProducerRecord is accessible at the same time as the RecordMetadata
>
> On Mon, Aug 26, 2019 at 10:26 AM Renuka M  wrote:
>
> > Hi Gwen,
> >
> > 1.We are not doing any changes on the broker side. This change is only on
> > Kafka clients library.
> > 2. RecordMetaData is created by client library while appending record to
> > ProducerBatch where offset alone returned by broker. Here we are adding
> > headers to RecordMetaData while creating FutureRecordMetaData to create
> > context between record and its metadata. I have updated the snippet in
> KIP
> > proposed changes in step 3.
> > 3. As we mentioned in alternatives, client side we can link record and
> its
> > metadata using callback, but Interceptors having same RecordMetadata will
> > not have context on for which record this MetaData belongs to. To fill
> that
> > Gap, we are proposing these changes.
> > Please let us know if we are not clear.
> >
> > Thanks
> > Renuka M
> >
> >
> >
> >
> > On Fri, Aug 23, 2019 at 7:08 PM Gwen Shapira  wrote:
> >
> > > I am afraid I don't understand the proposal. The RecordMetadata is
> > > information returned from the broker regarding the record. The
> > > producer already has the record (including the headers), so why would
> > > the broker need to send the headers back as part of the metadata?
> > >
> > > On Fri, Aug 23, 2019 at 4:22 PM Renuka M 
> wrote:
> > > >
> > > > Hi All,
> > > >
> > > > I am starting this thread to discuss
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-512%3AAdding+headers+to+RecordMetaData
> > > > .
> > > >
> > > > Please provide the feedback.
> > > >
> > > > Thanks
> > > > Renuka M
> > >
> > >
> > >
> > > --
> > > Gwen Shapira
> > > Product Manager | Confluent
> > > 650.450.2760 | @gwenshap
> > > Follow us: Twitter | blog
> > >
> >
>

[jira] [Reopened] (KAFKA-8800) Flaky Test SaslScramSslEndToEndAuthorizationTest#testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl

[Anna Povzner (Jira)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-8800?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Anna Povzner reopened KAFKA-8800:
-
  Assignee: Anastasia Vela  (was: Lee Dongjin)

> Flaky Test 
> SaslScramSslEndToEndAuthorizationTest#testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl
> --
>
> Key: KAFKA-8800
> URL: https://issues.apache.org/jira/browse/KAFKA-8800
> Project: Kafka
>  Issue Type: Bug
>  Components: core, security, unit tests
>Affects Versions: 2.4.0
>Reporter: Matthias J. Sax
>Assignee: Anastasia Vela
>Priority: Critical
>  Labels: flaky-test
> Fix For: 2.4.0, 2.3.1
>
>
> [https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/6956/testReport/junit/kafka.api/SaslScramSslEndToEndAuthorizationTest/testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl/]
> {quote}org.scalatest.exceptions.TestFailedException: Consumed 0 records 
> before timeout instead of the expected 1 records at 
> org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:530) at 
> org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:529) 
> at 
> org.scalatest.Assertions$.newAssertionFailedException(Assertions.scala:1389) 
> at org.scalatest.Assertions.fail(Assertions.scala:1091) at 
> org.scalatest.Assertions.fail$(Assertions.scala:1087) at 
> org.scalatest.Assertions$.fail(Assertions.scala:1389) at 
> kafka.utils.TestUtils$.waitUntilTrue(TestUtils.scala:822) at 
> kafka.utils.TestUtils$.pollRecordsUntilTrue(TestUtils.scala:781) at 
> kafka.utils.TestUtils$.pollUntilAtLeastNumRecords(TestUtils.scala:1312) at 
> kafka.utils.TestUtils$.consumeRecords(TestUtils.scala:1320) at 
> kafka.api.EndToEndAuthorizationTest.consumeRecords(EndToEndAuthorizationTest.scala:522)
>  at 
> kafka.api.EndToEndAuthorizationTest.testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl(EndToEndAuthorizationTest.scala:361){quote}



--
This message was sent by Atlassian Jira
(v8.3.2#803003)

Re: [DISCUSS] KIP-495: Dynamically Adjust Log Levels in Connect

[Jason Gustafson]

Hi Arjun,

>From a high level, I feel like we are making light of the JMX api because
it's convenient and the broker already has it. Personally I would take the
broker out of the picture. The JMX endpoint is not something we were happy
with, hence KIP-412. Ultimately I think we will deprecate and remove it and
there's no point trying to standardize on a deprecated mechanism. Thinking
just about connect, we already have an HTTP endpoint. The default position
should be to add new APIs to it rather than introducing other mechanisms.
The fewer ways you have to interact with a system, the better, right?

I think the main argument against a REST endpoint is basically that
adjusting log levels is an administrative operation and connect is lacking
an authorization framework to enforce administrative access. The same
argument applies to JMX, but it has the benefit that you can specify
different credentials and it is easier to isolate since it is running on a
separate port. As you suggested, I think the same benefits could be
achieved by having a separate /admin endpoint which is exposed (perhaps
optionally) on another listener. This is a pretty standard pattern. If
memory serves, dropwizard has something like this out of the box. We should
think hard whether there are additional administrative capabilities that we
would ultimately need. The answer is probably yes, so unless we want to
double down on JMX, it might be worth thinking through the implications of
an admin endpoint now so that we're not left with odd compatibility baggage
in the future.

Thanks,
Jason




On Fri, Aug 23, 2019 at 5:38 PM Arjun Satish  wrote:

> Jason,
>
> Thanks for your comments!
>
> I understand the usability issues with JMX that you mention. But it was
> chosen for the following reasons:
>
> 1. Cross-cutting functionality across different components (Kafka brokers,
> Connect workers and even with Streams jobs). If we go down the REST route,
> then brokers don't get this feature.
> 2. Adding this to existing REST servers adds the whole-or-nothing problem.
> It's hard to disable an endpoint if the functionality is not desired or
> needs to be protected from users (Connect doesn't have ACLs which makes
> this even harder to manage). Adding endpoints to different listeners makes
> configuring Connect harder (and it's already a hard problem as it is). A
> lot of the existing functionality there is driven around the connector data
> model (connectors, plugins, their statuses and so on). Adding an '/admin'
> endpoint may be a way to go, but that has tremendous implications (we are
> effectively adding an administration endpoint similar to the admin one in
> brokers), and probably requires a KIP of its own with discussions catered
> around just that.
> 3. JMX is currently AK's default way to report metrics and perform other
> operations. Changing log levels is typically a system level/admin
> operation, and fits better there, instead of REST APIs (which is more user
> facing).
>
> Having said that, I'm happy to consider alternatives. JMX seemed to be the
> lowest hanging fruit. But if there are better ideas, we can consider them.
> At the end of the day, when we download and run Kafka, there should be one
> way to achieve the same functionality among its components.
>
> Finally, I hope I didn't convey that we are reverting/changing the changes
> made in KIP-412. The proposed changes would be an addition to it. It will
> give brokers multiple ways of changing log levels. and there is still a
> consistent way of achieving cross component goals of the KIP.
>
> Best,
>
>
> On Fri, Aug 23, 2019 at 4:12 PM Jason Gustafson 
> wrote:
>
> > Let me elaborate a little bit. We made the decision early on for Connect
> to
> > use HTTP instead of Kafka's custom RPC protocol. In exchange for losing
> > some hygienic consistency with Kafka, we took easier integration with
> > management tools. The scope of the connect REST APIs is really managing
> the
> > connect cluster. It has endpoints for creating connectors, changing
> > configs, seeing their health, etc. Doesn't debugging fit in with that? I
> am
> > not sure I see why we would treat this as an exceptional case.
> >
> > I personally see JMX as a necessary evil in Kafka because most metrics
> > agents have native support. But it is particularly painful when it comes
> to
> > use as an RPC mechanism. This was the central motivation behind KIP-412,
> > which makes it very odd to see a new proposal which suggests
> standardizing
> > on JMX for log level adjustment. I actually see this as something we'd
> want
> > to eventually turn off in Kafka. Now that we have a proper API with
> support
> > in the AdminClient, we can deprecate and eventually remove the JMX
> > endpoint.
> >
> > Thanks,
> > Jason
> >
> > On Fri, Aug 23, 2019 at 10:49 AM Jason Gustafson 
> > wrote:
> >
> > > Hi Arjun,
> > >
> > > Thanks for the KIP. Do we really need a JMX-based API? Is there
> literally
> > > anyone in the world that 

[jira] [Created] (KAFKA-8837) KafkaMetricReporterClusterIdTest may not shutdown ZooKeeperTestHarness

[Anna Povzner (Jira)]

Anna Povzner created KAFKA-8837:
---

 Summary: KafkaMetricReporterClusterIdTest may not shutdown 
ZooKeeperTestHarness
 Key: KAFKA-8837
 URL: https://issues.apache.org/jira/browse/KAFKA-8837
 Project: Kafka
  Issue Type: Bug
  Components: core
Reporter: Anna Povzner
Assignee: Anastasia Vela


@After method in KafkaMetricReporterClusterIdTest calls  
`TestUtils.verifyNonDaemonThreadsStatus(this.getClass.getName)` before it calls 
tearDown on ZooKeeperTestHarness (which shut downs ZK and zk client). If 
verifyNonDaemonThreadsStatus asserts, the rest of the resources will not get 
cleaned up.

We should move `TestUtils.verifyNonDaemonThreadsStatus(this.getClass.getName)` 
to the end of `tearDown()`. However, would also be good to prevent people using 
this method in tear down similarly in the future. Maybe just adding a comment 
would help here.



--
This message was sent by Atlassian Jira
(v8.3.2#803003)

Build failed in Jenkins: kafka-trunk-jdk11 #775

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-8412: Fix nullpointer exception thrown on flushing before 
closing

[harsha] KAFKA-8669: Add security providers in kafka security config (#7090)

--
[...truncated 2.60 MB...]

org.apache.kafka.connect.transforms.RegexRouterTest > staticReplacement STARTED

org.apache.kafka.connect.transforms.RegexRouterTest > staticReplacement PASSED

org.apache.kafka.connect.transforms.ReplaceFieldTest > withSchema STARTED

org.apache.kafka.connect.transforms.ReplaceFieldTest > withSchema PASSED

org.apache.kafka.connect.transforms.ReplaceFieldTest > schemaless STARTED

org.apache.kafka.connect.transforms.ReplaceFieldTest > schemaless PASSED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > schemaNameUpdate 
STARTED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > schemaNameUpdate 
PASSED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > 
schemaNameAndVersionUpdateWithStruct STARTED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > 
schemaNameAndVersionUpdateWithStruct PASSED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > 
updateSchemaOfStruct STARTED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > 
updateSchemaOfStruct PASSED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > 
schemaNameAndVersionUpdate STARTED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > 
schemaNameAndVersionUpdate PASSED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > updateSchemaOfNull 
STARTED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > updateSchemaOfNull 
PASSED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > schemaVersionUpdate 
STARTED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > schemaVersionUpdate 
PASSED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > 
updateSchemaOfNonStruct STARTED

org.apache.kafka.connect.transforms.SetSchemaMetadataTest > 
updateSchemaOfNonStruct PASSED

org.apache.kafka.connect.transforms.HoistFieldTest > withSchema STARTED

org.apache.kafka.connect.transforms.HoistFieldTest > withSchema PASSED

org.apache.kafka.connect.transforms.HoistFieldTest > schemaless STARTED

org.apache.kafka.connect.transforms.HoistFieldTest > schemaless PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testSchemalessFieldConversion STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testSchemalessFieldConversion PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testConfigInvalidTargetType STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testConfigInvalidTargetType PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testSchemalessStringToTimestamp STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testSchemalessStringToTimestamp PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > testKey STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > testKey PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testSchemalessDateToTimestamp STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testSchemalessDateToTimestamp PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testSchemalessTimestampToString STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testSchemalessTimestampToString PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testSchemalessTimeToTimestamp STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testSchemalessTimeToTimestamp PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaTimestampToDate STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaTimestampToDate PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaTimestampToTime STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaTimestampToTime PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaTimestampToUnix STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaTimestampToUnix PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaNullValueToTimestamp STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaNullValueToTimestamp PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaNullValueToDate STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaNullValueToDate PASSED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaNullValueToTime STARTED

org.apache.kafka.connect.transforms.TimestampConverterTest > 
testWithSchemaNullValueToTime PASSED

Build failed in Jenkins: kafka-2.3-jdk8 #96

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-8412: Fix nullpointer exception thrown on flushing before 
closing

--
[...truncated 58.39 KB...]
:821:
 constructor PartitionData in class PartitionData is deprecated: see 
corresponding Javadoc for more information.
  new ListOffsetResponse.PartitionData(Errors.TOPIC_AUTHORIZATION_FAILED, 
List[JLong]().asJava)
  ^
:829:
 value maxNumOffsets in class PartitionData is deprecated: see corresponding 
Javadoc for more information.
  maxNumOffsets = partitionData.maxNumOffsets,
^
:832:
 constructor PartitionData in class PartitionData is deprecated: see 
corresponding Javadoc for more information.
(topicPartition, new ListOffsetResponse.PartitionData(Errors.NONE, 
offsets.map(JLong.valueOf).asJava))
 ^
:841:
 constructor PartitionData in class PartitionData is deprecated: see 
corresponding Javadoc for more information.
  (topicPartition, new 
ListOffsetResponse.PartitionData(Errors.forException(e), List[JLong]().asJava))
   ^
:844:
 constructor PartitionData in class PartitionData is deprecated: see 
corresponding Javadoc for more information.
  (topicPartition, new 
ListOffsetResponse.PartitionData(Errors.forException(e), List[JLong]().asJava))
   ^
:231:
 value DEFAULT_SASL_ENABLED_MECHANISMS in class SaslConfigs is deprecated: see 
corresponding Javadoc for more information.
  val SaslEnabledMechanisms = SaslConfigs.DEFAULT_SASL_ENABLED_MECHANISMS
  ^
:235:
 value offsets in class PartitionData is deprecated: see corresponding Javadoc 
for more information.
  responsePartitionData.offsets.get(0)
^
:573:
 method checksum in class ConsumerRecord is deprecated: see corresponding 
Javadoc for more information.
output.println(topicStr + "checksum:" + consumerRecord.checksum)
   ^
:197:
 class BaseConsumerRecord in package consumer is deprecated (since 0.11.0.0): 
This class has been deprecated and will be removed in a future release. Please 
use org.apache.kafka.clients.consumer.ConsumerRecord instead.
private def toBaseConsumerRecord(record: ConsumerRecord[Array[Byte], 
Array[Byte]]): BaseConsumerRecord =

^
:198:
 class BaseConsumerRecord in package consumer is deprecated (since 0.11.0.0): 
This class has been deprecated and will be removed in a future release. Please 
use org.apache.kafka.clients.consumer.ConsumerRecord instead.
  BaseConsumerRecord(record.topic,
  ^
:417:
 class BaseConsumerRecord in package consumer is deprecated (since 0.11.0.0): 
This class has been deprecated and will be removed in a future release. Please 
use org.apache.kafka.clients.consumer.ConsumerRecord instead.
def handle(record: BaseConsumerRecord): 
util.List[ProducerRecord[Array[Byte], Array[Byte]]]
   ^
:421:
 class BaseConsumerRecord in package consumer is deprecated (since 0.11.0.0): 
This class has been deprecated and will be removed in a future release. Please 
use org.apache.kafka.clients.consumer.ConsumerRecord instead.
override def handle(record: BaseConsumerRecord): 
util.List[ProducerRecord[Array[Byte], Array[Byte]]] = {
^
34 warnings found
Note: 

 uses unchecked or unsafe operations.
Note: Recompile with -Xlint:unchecked for details.

> Task 

Re: [ DISCUSS ] KIP-512:Adding headers to RecordMetaData

[Eric Azama]

Have you considered adding a new onAcknowledgement method to the
ProducerInterceptor with the signature onAcknowledgement(RecordMetadata
metadata, Exception exception, ProducerRecord record)? I would also
consider adding this to Producer Callbacks as well, since linking a
Callback to a specific record currently requires creating a new Callback
for every ProducerRecord sent.

This seems like a more robust strategy compared to using Headers. Headers
don't necessarily contain anything that connects them to the original
ProducerRecord, and forcibly including information in the Headers seems
like unnecessary bloat. If your goal is to link a RecordMetadata to a
specific ProducerRecord, it seems simpler to make sure the original
ProducerRecord is accessible at the same time as the RecordMetadata

On Mon, Aug 26, 2019 at 10:26 AM Renuka M  wrote:

> Hi Gwen,
>
> 1.We are not doing any changes on the broker side. This change is only on
> Kafka clients library.
> 2. RecordMetaData is created by client library while appending record to
> ProducerBatch where offset alone returned by broker. Here we are adding
> headers to RecordMetaData while creating FutureRecordMetaData to create
> context between record and its metadata. I have updated the snippet in KIP
> proposed changes in step 3.
> 3. As we mentioned in alternatives, client side we can link record and its
> metadata using callback, but Interceptors having same RecordMetadata will
> not have context on for which record this MetaData belongs to. To fill that
> Gap, we are proposing these changes.
> Please let us know if we are not clear.
>
> Thanks
> Renuka M
>
>
>
>
> On Fri, Aug 23, 2019 at 7:08 PM Gwen Shapira  wrote:
>
> > I am afraid I don't understand the proposal. The RecordMetadata is
> > information returned from the broker regarding the record. The
> > producer already has the record (including the headers), so why would
> > the broker need to send the headers back as part of the metadata?
> >
> > On Fri, Aug 23, 2019 at 4:22 PM Renuka M  wrote:
> > >
> > > Hi All,
> > >
> > > I am starting this thread to discuss
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-512%3AAdding+headers+to+RecordMetaData
> > > .
> > >
> > > Please provide the feedback.
> > >
> > > Thanks
> > > Renuka M
> >
> >
> >
> > --
> > Gwen Shapira
> > Product Manager | Confluent
> > 650.450.2760 | @gwenshap
> > Follow us: Twitter | blog
> >
>

Re: [ DISCUSS ] KIP-512:Adding headers to RecordMetaData

[Renuka M]

Hi Gwen,

1.We are not doing any changes on the broker side. This change is only on
Kafka clients library.
2. RecordMetaData is created by client library while appending record to
ProducerBatch where offset alone returned by broker. Here we are adding
headers to RecordMetaData while creating FutureRecordMetaData to create
context between record and its metadata. I have updated the snippet in KIP
proposed changes in step 3.
3. As we mentioned in alternatives, client side we can link record and its
metadata using callback, but Interceptors having same RecordMetadata will
not have context on for which record this MetaData belongs to. To fill that
Gap, we are proposing these changes.
Please let us know if we are not clear.

Thanks
Renuka M




On Fri, Aug 23, 2019 at 7:08 PM Gwen Shapira  wrote:

> I am afraid I don't understand the proposal. The RecordMetadata is
> information returned from the broker regarding the record. The
> producer already has the record (including the headers), so why would
> the broker need to send the headers back as part of the metadata?
>
> On Fri, Aug 23, 2019 at 4:22 PM Renuka M  wrote:
> >
> > Hi All,
> >
> > I am starting this thread to discuss
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-512%3AAdding+headers+to+RecordMetaData
> > .
> >
> > Please provide the feedback.
> >
> > Thanks
> > Renuka M
>
>
>
> --
> Gwen Shapira
> Product Manager | Confluent
> 650.450.2760 | @gwenshap
> Follow us: Twitter | blog
>

Build failed in Jenkins: kafka-2.1-jdk8 #226

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-8412: Fix nullpointer exception thrown on flushing before 
closing

--
[...truncated 91.06 KB...]
return  valueSerde != null ? valueSerde.deserializer() : null;
^
  required: Deserializer
  found:Deserializer
  where V is a type-variable:
V extends Object declared in class OptimizableRepartitionNode
:78:
 warning: [unchecked] unchecked conversion
final Serializer keySerializer = keySerde != null ? 
keySerde.serializer() : null;

  ^
  required: Serializer
  found:Serializer
  where K is a type-variable:
K extends Object declared in class OptimizableRepartitionNode
:79:
 warning: [unchecked] unchecked conversion
final Deserializer keyDeserializer = keySerde != null ? 
keySerde.deserializer() : null;

^
  required: Deserializer
  found:Deserializer
  where K is a type-variable:
K extends Object declared in class OptimizableRepartitionNode
:80:
 warning: [unchecked] unchecked cast
topologyBuilder.addGlobalStore((StoreBuilder) 
storeBuilder,
 ^
  required: StoreBuilder
  found:StoreBuilder
  where S is a type-variable:
S extends StateStore declared in class TableSourceNode
:133:
 warning: [unchecked] unchecked conversion
this.consumedInternal = consumedInternal;
^
  required: ConsumedInternal
  found:ConsumedInternal
  where K,V are type-variables:
K extends Object declared in class TableSourceNodeBuilder
V extends Object declared in class TableSourceNodeBuilder
:61:
 warning: [unchecked] unchecked method invocation: constructor  in class 
WindowedStreamPartitioner is applied to given types
final StreamPartitioner windowedPartitioner = 
(StreamPartitioner) new WindowedStreamPartitioner((WindowedSerializer) keySerializer);

  ^
  required: WindowedSerializer
  found: WindowedSerializer
  where K is a type-variable:
K extends Object declared in class WindowedStreamPartitioner
:61:
 warning: [unchecked] unchecked conversion
final StreamPartitioner windowedPartitioner = 
(StreamPartitioner) new WindowedStreamPartitioner((WindowedSerializer) keySerializer);

   ^
  required: WindowedSerializer
  found:WindowedSerializer
  where K is a type-variable:
K extends Object declared in class WindowedStreamPartitioner
:61:
 warning: [unchecked] unchecked cast
final StreamPartitioner windowedPartitioner = 
(StreamPartitioner) new WindowedStreamPartitioner((WindowedSerializer) keySerializer);

  ^
  required: StreamPartitioner
  found:WindowedStreamPartitioner
  where V,K are type-variables:
V extends Object declared in class StreamSinkNode
K extends Object declared in class StreamSinkNode
:88:
 warning: [unchecked] unchecked method invocation: constructor  in class 
KeyValueStoreMaterializer is applied to given types
= new 
KeyValueStoreMaterializer<>(materializedInternal).materialize();
  ^
  required: MaterializedInternal>
  found: MaterializedInternal
  where K,V are type-variables:
K extends Object declared in class 

[jira] [Created] (KAFKA-8836) Add inter-broker protocol to alter ISR

[Jason Gustafson (Jira)]

Jason Gustafson created KAFKA-8836:
--

 Summary: Add inter-broker protocol to alter ISR
 Key: KAFKA-8836
 URL: https://issues.apache.org/jira/browse/KAFKA-8836
 Project: Kafka
  Issue Type: Improvement
Reporter: Jason Gustafson
Assignee: Jason Gustafson


This tracks the implementation of KIP-497: 
[https://cwiki.apache.org/confluence/display/KAFKA/KIP-497%3A+Add+inter-broker+API+to+alter+ISR].
 Likely this will be broken down into sub-tasks.



--
This message was sent by Atlassian Jira
(v8.3.2#803003)

Build failed in Jenkins: kafka-2.2-jdk8 #162

[Apache Jenkins Server]

See 


Changes:

[wangguoz] KAFKA-8412: Fix nullpointer exception thrown on flushing before 
closing

--
Started by an SCM change
[EnvInject] - Loading node environment variables.
Building remotely on H27 (ubuntu xenial) in workspace 

[WS-CLEANUP] Deleting project workspace...
[WS-CLEANUP] Deferred wipeout is used...
No credentials specified
Cloning the remote Git repository
Cloning repository https://github.com/apache/kafka.git
 > git init  # timeout=10
Fetching upstream changes from https://github.com/apache/kafka.git
 > git --version # timeout=10
 > git fetch --tags --progress https://github.com/apache/kafka.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git config remote.origin.url https://github.com/apache/kafka.git # timeout=10
 > git config --add remote.origin.fetch +refs/heads/*:refs/remotes/origin/* # 
 > timeout=10
 > git config remote.origin.url https://github.com/apache/kafka.git # timeout=10
Fetching upstream changes from https://github.com/apache/kafka.git
 > git fetch --tags --progress https://github.com/apache/kafka.git 
 > +refs/heads/*:refs/remotes/origin/*
 > git rev-parse refs/remotes/origin/2.2^{commit} # timeout=10
 > git rev-parse refs/remotes/origin/origin/2.2^{commit} # timeout=10
Checking out Revision b29dde6f6eabcf2d658f1d4055a55d139783eb4c 
(refs/remotes/origin/2.2)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f b29dde6f6eabcf2d658f1d4055a55d139783eb4c
Commit message: "KAFKA-8412: Fix nullpointer exception thrown on flushing 
before closing producers (#7207)"
 > git rev-list --no-walk d6c7b75f8d609b3679b291418f584eb97563b48c # timeout=10
Setting GRADLE_4_8_1_HOME=/home/jenkins/tools/gradle/4.8.1
Setting GRADLE_4_8_1_HOME=/home/jenkins/tools/gradle/4.8.1
[kafka-2.2-jdk8] $ /bin/bash -xe /tmp/jenkins3773796488198048544.sh
+ rm -rf 
+ /home/jenkins/tools/gradle/4.8.1/bin/gradle
/tmp/jenkins3773796488198048544.sh: line 4: 
/home/jenkins/tools/gradle/4.8.1/bin/gradle: No such file or directory
Build step 'Execute shell' marked build as failure
[FINDBUGS] Collecting findbugs analysis files...
Setting GRADLE_4_8_1_HOME=/home/jenkins/tools/gradle/4.8.1
[FINDBUGS] Searching for all files in 
 that match the pattern 
**/build/reports/findbugs/*.xml
[FINDBUGS] No files found. Configuration error?
Setting GRADLE_4_8_1_HOME=/home/jenkins/tools/gradle/4.8.1
No credentials specified
Setting GRADLE_4_8_1_HOME=/home/jenkins/tools/gradle/4.8.1
 Using GitBlamer to create author and commit information for all 
warnings.
 GIT_COMMIT=b29dde6f6eabcf2d658f1d4055a55d139783eb4c, 
workspace=
[FINDBUGS] Computing warning deltas based on reference build #161
Recording test results
Setting GRADLE_4_8_1_HOME=/home/jenkins/tools/gradle/4.8.1
ERROR: Step ?Publish JUnit test result report? failed: No test report files 
were found. Configuration error?
Setting GRADLE_4_8_1_HOME=/home/jenkins/tools/gradle/4.8.1
Not sending mail to unregistered user wangg...@gmail.com

[jira] [Created] (KAFKA-8835) Update documentation for URP changes in KIP-352

[Jason Gustafson (Jira)]

Jason Gustafson created KAFKA-8835:
--

 Summary: Update documentation for URP changes in KIP-352
 Key: KAFKA-8835
 URL: https://issues.apache.org/jira/browse/KAFKA-8835
 Project: Kafka
  Issue Type: Improvement
Reporter: Jason Gustafson


This Jira covers any doc changes needed for the changes to URP semantics in 
KIP-352: 
[https://cwiki.apache.org/confluence/display/KAFKA/KIP-352%3A+Distinguish+URPs+caused+by+reassignment].



--
This message was sent by Atlassian Jira
(v8.3.2#803003)

[jira] [Created] (KAFKA-8834) Distinguish URPs caused by reassignment plus other metrics

[Jason Gustafson (Jira)]

Jason Gustafson created KAFKA-8834:
--

 Summary: Distinguish URPs caused by reassignment plus other metrics
 Key: KAFKA-8834
 URL: https://issues.apache.org/jira/browse/KAFKA-8834
 Project: Kafka
  Issue Type: Improvement
Reporter: Jason Gustafson
Assignee: Jason Gustafson


This Jira tracks implementation of KIP-352: 
[https://cwiki.apache.org/confluence/display/KAFKA/KIP-352%3A+Distinguish+URPs+caused+by+reassignment].



--
This message was sent by Atlassian Jira
(v8.3.2#803003)

[jira] [Resolved] (KAFKA-8412) Still a nullpointer exception thrown on shutdown while flushing before closing producers

[Guozhang Wang (Jira)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-8412?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

Guozhang Wang resolved KAFKA-8412.
--
Resolution: Fixed

> Still a nullpointer exception thrown on shutdown while flushing before 
> closing producers
> 
>
> Key: KAFKA-8412
> URL: https://issues.apache.org/jira/browse/KAFKA-8412
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 2.1.1
>Reporter: Sebastiaan
>Assignee: Chris Pettitt
>Priority: Minor
> Fix For: 2.1.2, 2.2.2, 2.4.0, 2.3.1
>
>
> I found a closed issue and replied there but decided to open one myself 
> because although they're related they're slightly different. The original 
> issue is at https://issues.apache.org/jira/browse/KAFKA-7678
> The fix there has been to implement a null check around closing a producer 
> because in some cases the producer is already null there (has been closed 
> already)
> In version 2.1.1 we are getting a very similar exception, but in the 'flush' 
> method that is called pre-close. This is in the log:
> {code:java}
> message: stream-thread 
> [webhook-poster-7034dbb0-7423-476b-98f3-d18db675d6d6-StreamThread-1] Failed 
> while closing StreamTask 1_26 due to the following error:
> logger_name: org.apache.kafka.streams.processor.internals.AssignedStreamsTasks
> java.lang.NullPointerException: null
>     at 
> org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:245)
>     at 
> org.apache.kafka.streams.processor.internals.StreamTask.flushState(StreamTask.java:493)
>     at 
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:443)
>     at 
> org.apache.kafka.streams.processor.internals.StreamTask.suspend(StreamTask.java:568)
>     at 
> org.apache.kafka.streams.processor.internals.StreamTask.close(StreamTask.java:691)
>     at 
> org.apache.kafka.streams.processor.internals.AssignedTasks.close(AssignedTasks.java:397)
>     at 
> org.apache.kafka.streams.processor.internals.TaskManager.shutdown(TaskManager.java:260)
>     at 
> org.apache.kafka.streams.processor.internals.StreamThread.completeShutdown(StreamThread.java:1181)
>     at 
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:758){code}
> Followed by:
>  
> {code:java}
> message: task [1_26] Could not close task due to the following error:
> logger_name: org.apache.kafka.streams.processor.internals.StreamTask
> java.lang.NullPointerException: null
>     at 
> org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:245)
>     at 
> org.apache.kafka.streams.processor.internals.StreamTask.flushState(StreamTask.java:493)
>     at 
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:443)
>     at 
> org.apache.kafka.streams.processor.internals.StreamTask.suspend(StreamTask.java:568)
>     at 
> org.apache.kafka.streams.processor.internals.StreamTask.close(StreamTask.java:691)
>     at 
> org.apache.kafka.streams.processor.internals.AssignedTasks.close(AssignedTasks.java:397)
>     at 
> org.apache.kafka.streams.processor.internals.TaskManager.shutdown(TaskManager.java:260)
>     at 
> org.apache.kafka.streams.processor.internals.StreamThread.completeShutdown(StreamThread.java:1181)
>     at 
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:758){code}
> If I look at the source code at this point, I see a nice null check in the 
> close method, but not in the flush method that is called just before that:
> {code:java}
> public void flush() {
>     this.log.debug("Flushing producer");
>     this.producer.flush();
>     this.checkForException();
> }
> public void close() {
>     this.log.debug("Closing producer");
>     if (this.producer != null) {
>     this.producer.close();
>     this.producer = null;
>     }
>     this.checkForException();
> }{code}
> Seems to my (ignorant) eye that the flush method should also be wrapped in a 
> null check in the same way as has been done for close.



--
This message was sent by Atlassian Jira
(v8.3.2#803003)

Re: [DISCUSS] KIP-481: SerDe Improvements for Connect Decimal type in JSON

[Almog Gavra]

Thanks for the feedback Randall! I have updated the KIP with the following
edits:

* Updated the reference from "producer" to "source" (I had missed that one!)
* Changed the config from "json.decimal.serialization.format" to
"decimal.format"
* Clarified case sensitivity
* Clarified the proposed changes to note that deserialization is not
affected by the config
* Clarified the changes in JsonConverter to handle deserialization (see my
third bullet below)
* Added a clear migration plan and simplified compatibility

Here are also some clarifications based on your comments.

* I think "json" has limited value in the configuration name. If put in a
top-level worker config, it clarifies that it only affects connectors using
the JsonConverter. I have opted for your suggestion and dropped it.
* I think "serialization" has limited value in the configuration name. If
we ever want to introduce "deserialization" configurations, there will be
asymmetry in the configuration names. I have opted for your suggestion and
dropped it.
* The JsonConverter will not "always look for numbers". The converter will
receive from the Jackson Object Mapper either a NumericNode containing a
big decimal or a BinaryNode containing a btye[]. Based on the type of this
node, it will convert the value to a BigDecimal appropriately (or any other
Connect java type based on the schema).
* "the ... JsonDeserializer are not affected" is not exactly true, but
semantically correct. See the note in the KIP about defaulting floating
points to BigDecimal to avoid precision loss.
* "The resulting application, however, may need to handle a wider range of
numeric values." Unless I misunderstand what you're saying, I don't think
this is correct. The resulting application will still receive exactly the
same Connect data object from the JsonConverter as it was before - only the
SerDe layer is affected.

Cheers,
Almog

On Sun, Aug 25, 2019 at 4:28 PM Randall Hauch  wrote:

> Thanks for all the work, Almog.
>
> For the most part, I think this KIP will be a great improvement, and IMO is
> almost ready to go. However, I do have a few suggestions that affect the
> wording more than the intent.
>
> First, the name of the `json.decimal.serialization.format` property is
> pretty long, especially when it is prefixed in the Worker config or in a
> connector config as `key.converer.json.decimal.serialization.format` or
> `value.converter.json.decimal.serialization.format` . Have you considered a
> shorter config property name, such as maybe `decimal.format`? Is there any
> benefit to include "json" and "serialization" in the property name? Also,
> we should be clear that the value will not be case sensitive (e.g.,
> "numeric" and "NUMERIC" would be equivalent), to keep in alignment with
> other enumeration literals in Connect configurations. The goal should be
> simple
>
> Second, the Motivation section, has the following sentence:
>
> "A new configuration for producers json.decimal.serialization.format will
> be introduced to the JsonConverter configuration to help control whether
> source converters will serialize decimals in numeric or binary formats."
>
>
> I agree with an earlier comment from Konstantine that "producers" here is
> distracting and does not mirror the normal definition of "producers" within
> the Kafka context. I suggest rephrasing this to something like
>
> "Introduce to the JsonConverter a new configuration property named
> json.decimal.serialization.format to control whether source converters will
> serialize decimals in numeric or binary formats."
>
>
> Third, the KIP should be more clear about whether the
> `json.decimal.serialization.format` setting does or does not affect
> deserialization? IIUC, the deserialization logic will always look for JSON
> numbers, and will always use the Schema to define whether it should convert
> the value to a different number type. Is that a fair statement?
>
> Fourth, the JsonSerializer and JsonDeserializer are not affected, yet are
> still compatible with the old and new behavior. Because the primary purpose
> of this new setting is to define how Connect DECIMAL logical type values
> are serialized in JSON documents, the JsonDeserializer will still be able
> to deserialize the JSON document correctly. The resulting application,
> however, may need to handle a wider range of numeric values.
>
> Fifth, the Compatibility section seems more complicated than perhaps it
> needs to be, maybe because it seems to distinguish between upgrading and
> setting the decimal serialization format. Maybe it would be sufficient to
> simply emphasize that all of the sink connectors (or consumer applications)
> using the JsonConverter with
> the `json.decimal.serialization.format=NUMERIC` setting consuming records
> from a set of topics be upgraded and changed *before* any of the source
> connectors (or other producer applications) using the JsonConverter to
> serialize records are changed to use
> the 

Re: [VOTE] KIP-352: Distinguish URPs caused by reassignment

[Jason Gustafson]

Closing this vote. The final result is +9 with 4 binding votes.

@Satish Sorry, I missed your question above. Good point about updating
documentation. I will create a separate jira to make sure this gets done.

-Jason

On Fri, Aug 23, 2019 at 11:23 AM Jason Gustafson  wrote:

> Thanks Stan, good catch. I have updated the KIP. I will plan to close the
> vote Monday if there are no objections.
>
> -Jason
>
> On Fri, Aug 23, 2019 at 11:14 AM Colin McCabe  wrote:
>
>> On Fri, Aug 23, 2019, at 11:08, Stanislav Kozlovski wrote:
>> > Thanks for the KIP, this is very helpful
>> >
>> > I had an offline discussion with Jason and we discussed the semantics of
>> > the underMinIsr/atMinIsr metrics. The current proposal would expose a
>> gap
>> > where we could report URP but no MinIsr.
>> > A brief example:
>> > original replica set = [0,1,2]
>> > new replica set = [3,4,5]
>> > isr = [0, 3, 4]
>> > config.minIsr = 3
>> >
>> > As the KIP said
>> > > In other words, we will subtract the AddingReplica from both the total
>> > replicas and the current ISR when determining URP satisfaction.
>> > We would report URP=2 (1 and 2 are not in ISR) but not underMinIsr, as
>> we
>> > have an ISR of 3.
>> > Technically, any produce requests with acks=all would succeed, so it
>> would
>> > be false to report `underMinIsr`. We thought it'd be good to keep both
>> > metrics consistent, so a new proposal is to use the following algorithm:
>> > ```
>> > isUrp == size(original replicas) - size(isr) > 0
>> > ```
>>
>> Hi Stan,
>>
>> That's a good point.  Basically we should regard the size of the original
>> replica set as the desired replication factor, and calculate the URPs based
>> on that.  +1 for this.  (I assume Jason will update the KIP...)
>>
>> best,
>> Colin
>>
>>
>> >
>> > Taking that into account, +1 from me! (non-binding)
>> >
>> > On Fri, Aug 23, 2019 at 7:00 PM Colin McCabe 
>> wrote:
>> >
>> > > +1 (binding).
>> > >
>> > > cheers,
>> > > Colin
>> > >
>> > > On Tue, Aug 20, 2019, at 10:55, Jason Gustafson wrote:
>> > > > Hi All,
>> > > >
>> > > > I'd like to start a vote on KIP-352, which is a follow-up to
>> KIP-455 to
>> > > > fix
>> > > > a long-known shortcoming of URP reporting and to improve
>> reassignment
>> > > > monitoring:
>> > > >
>> > >
>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-352%3A+Distinguish+URPs+caused+by+reassignment
>> > > > .
>> > > >
>> > > > Note that I have added one new metric following the discussion. It
>> seemed
>> > > > useful to have a lag metric for reassigning partitions.
>> > > >
>> > > > Thanks,
>> > > > Jason
>> > > >
>> > >
>> >
>> >
>> > --
>> > Best,
>> > Stanislav
>> >
>>
>

[jira] [Created] (KAFKA-8833) Metric for Broker Consumer Lag in size and time

[NIKHIL (Jira)]

NIKHIL created KAFKA-8833:
-

 Summary: Metric for Broker Consumer Lag in size and time
 Key: KAFKA-8833
 URL: https://issues.apache.org/jira/browse/KAFKA-8833
 Project: Kafka
  Issue Type: Improvement
  Components: metrics
Reporter: NIKHIL


In many use cases consumers are expected to fetch at the tail end of the log. 
The proposal is to have stats on consumer lag in terms for bytes as well as 
time. We will add broker level metrics aggregated across all partitions. Both 
metrics will be histogram.
 
- FetchLagTimeInMs: Histogram that will measure fetch log lag using the 
timestamp of the messages being fetched  
- FetchLagBytes: Histogram that will measure the fetch log lag by calculating 
the byte lag of the message being fetched
 



--
This message was sent by Atlassian Jira
(v8.3.2#803003)

Build failed in Jenkins: kafka-trunk-jdk11 #774

[Apache Jenkins Server]

See 


Changes:

[gwen] KAFKA-8391; Temporarily ignore flaky Connect rebalance integration tests

[rhauch] MINOR: Fix the doc of scheduled.rebalance.max.delay.ms config property

--
[...truncated 2.26 MB...]
java.lang.AssertionError: 
Expected: 
 but: was null
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:18)
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:6)
at 
org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest.shouldReduceSessionWindows(KStreamAggregationIntegrationTest.java:667)

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldReduce STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldReduce PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldAggregate STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldAggregate PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCount STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCount PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldGroupByKey STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldGroupByKey PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCountWithInternalStore STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCountWithInternalStore PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCountUnlimitedWindows STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCountUnlimitedWindows PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldReduceWindowed STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldReduceWindowed PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCountSessionWindows STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldCountSessionWindows PASSED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldAggregateWindowed STARTED

org.apache.kafka.streams.integration.KStreamAggregationIntegrationTest > 
shouldAggregateWindowed PASSED

org.apache.kafka.streams.integration.SmokeTestDriverIntegrationTest > 
shouldWorkWithRebalance STARTED
org.apache.kafka.streams.integration.SmokeTestDriverIntegrationTest.shouldWorkWithRebalance
 failed, log available in 


org.apache.kafka.streams.integration.SmokeTestDriverIntegrationTest > 
shouldWorkWithRebalance FAILED
java.lang.AssertionError: verifying tagg
fail: key=694 tagg=[ConsumerRecord(topic = tagg, partition = 0, leaderEpoch 
= 0, offset = 5, CreateTime = 1566824033327, serialized key size = 3, 
serialized value size = 8, headers = RecordHeaders(headers = [], isReadOnly = 
false), key = 694, value = 2)] expected=null
 outputEvents: [ConsumerRecord(topic = tagg, partition = 0, leaderEpoch 
= 0, offset = 5, CreateTime = 1566824033327, serialized key size = 3, 
serialized value size = 8, headers = RecordHeaders(headers = [], isReadOnly = 
false), key = 694, value = 2)]
verifying suppressed min-suppressed
verifying min-suppressed with 10 keys
verifying suppressed sws-suppressed
verifying min with 10 keys
verifying max with 10 keys
verifying dif with 10 keys
verifying sum with 10 keys
verifying cnt with 10 keys
verifying avg with 10 keys
at org.junit.Assert.fail(Assert.java:89)
at org.junit.Assert.assertTrue(Assert.java:42)
at 
org.apache.kafka.streams.integration.SmokeTestDriverIntegrationTest.shouldWorkWithRebalance(SmokeTestDriverIntegrationTest.java:137)

org.apache.kafka.streams.integration.KStreamAggregationDedupIntegrationTest > 
shouldReduce STARTED

org.apache.kafka.streams.integration.KStreamAggregationDedupIntegrationTest > 
shouldReduce PASSED

org.apache.kafka.streams.integration.KStreamAggregationDedupIntegrationTest > 
shouldGroupByKey STARTED

org.apache.kafka.streams.integration.KStreamAggregationDedupIntegrationTest > 
shouldGroupByKey PASSED

org.apache.kafka.streams.integration.KStreamAggregationDedupIntegrationTest > 
shouldReduceWindowed STARTED

org.apache.kafka.streams.integration.KStreamAggregationDedupIntegrationTest > 
shouldReduceWindowed PASSED

org.apache.kafka.streams.integration.SuppressionDurabilityIntegrationTest > 
shouldRecoverBufferAfterShutdown[0: eosEnabled=false] STARTED

Re: [DISCUSS] KIP-482: The Kafka Protocol should Support Optional Fields

[Magnus Edenhill]

Great KIP as always, Colin!

Some comments:

>  If the flexible versions are not specified, it is assumed that all
versions are flexible.

This is ambiguous, if a protocol-generator is pointed to an older Kafka
protocol specification
it can't know if the lack of flexibleVersions field means they're all
flexible, or flexibleVersions
is not supported at all.
I recommend requiring the flexibleVersions field if there are flexible
versions.


> They are serialized in ascending order of their tag.

I'm not opposed to it, but what's the reasoning for requiring the tag list
to be ordered?
And if it is a requirement, make the wording stronger ("required"),
otherwise remove it.


> All requests and responses will end with a tagged field buffer.  If there
are no tagged fields, this will only be a single zero byte.

Since the tagged fields can (or most likely will) contain information that
has bearing on the information in the request or response, it'd be better if
the tagged fields were inserted before the request/response body.
A streaming protocol parser would otherwise have to perform two passes of
the req/resp (worst case).


Addtional comments:

1)
With request/response tags we can now introduce request-level errors,
allowing the broker
to inform the client why a request was rejected before closing down the
connection.
E.g., an empty response body with tag 0 as the error code and tag 1 as the
error string.

2)
The protocol generator should error out if there are duplicate tags for a
context,
or if a context has tagged fields but no flexibleVersions, to help catch
such errors early.


3)
A semi-related ask:
there's currently some non-JSON comments in the protocol spec files,
typically version history.
If the information in the comment is important enough to be in the spec
file, it should be in
a proper JSON field (e.g., about), which makes sure it ends up in the
generated protocol docs instead of being filtered out.


/Magnus


Den mån 19 aug. 2019 kl 19:00 skrev Colin McCabe :

> Hi Satish,
>
> I wasn't originally going to propose supporting the struct type, although
> perhaps we could consider it.
>
> In general, supporting a struct containing an array has the same
> serialization issues as just supporting the array.
>
> Probably, we should just have a two-pass serialization mechanism where we
> calculate lengths first and then write out bytes.  If we do that, we can
> avoid having any restrictions on tagged fields vs. regular fields.  I will
> take a look at how complex this would be.
>
> best,
> Colin
>
>
> On Sun, Aug 18, 2019, at 22:27, Satish Duggana wrote:
> > Please read struct type as a complex record type in my earlier mail.
> > The complex type seems to be defined as Schema[1] in the protocol
> > types.
> >
> > 1.
> >
> https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/protocol/types/Schema.java#L27
> >
> >
> > On Mon, Aug 19, 2019 at 9:46 AM Satish Duggana 
> wrote:
> > >
> > > Sorry! Colin, I may not have been clear in my earlier query about
> > > optional field type restriction. It is mentioned in one of your
> > > replies "optional fields are serialized starting with their total
> > > length". This brings the question of whether optional fields support
> > > struct types (with or without array values). It seems struct types are
> > > currently not serialized with total length. I may be missing something
> > > here.
> > >
> > > Thanks,
> > > Satish.
> > >
> > >
> > > On Wed, Aug 14, 2019 at 8:03 AM Satish Duggana <
> satish.dugg...@gmail.com> wrote:
> > > >
> > > > Hi Colin,
> > > > Thanks for the KIP. Optional fields and var length encoding support
> is a great
> > > > improvement for the protocol.
> > > >
> > > > >>Optional fields can have any type, except that they cannot be
> arrays.
> > > > Note that the restriction against having tagged arrays is just to
> simplify
> > > > serialization.  We can relax this restriction in the future without
> changing
> > > > the protocol on the wire.
> > > >
> > > > Can an Optional field have a struct type which internally contains
> an array
> > > > field at any level?
> > > >
> > > > Thanks,
> > > > Satish.
> > > >
> > > >
> > > >
> > > > On Tue, Aug 13, 2019 at 11:49 PM David Jacot 
> wrote:
> > > > >
> > > > > Hi Colin,
> > > > >
> > > > > Thank you for the KIP! Things are well explained!. It is huge
> improvement
> > > > > for the Kafka protocol. I have few comments on the proposal:
> > > > >
> > > > > 1. The interleaved tag/length header sounds like a great
> optimisation as it
> > > > > would be shorter on average. The downside, as
> > > > > you already pointed out, is that it makes the decoding and the
> specs more
> > > > > complex. Personally, I would also favour using two
> > > > > vaints in this particular case to keep things simple.
> > > > >
> > > > > 2. As discussed, I wonder if it would make sense to extend to KIP
> to also
> > > > > support optional fields in the Record Header. I think
> > > > > that it 

Re: [DISCUSS] KIP-447: Producer scalability for exactly once semantics

[Boyang Chen]

Hey Guozhang and Jason,

I'm ok with either way. Thinking of Guozhang's approach, it is simpler to
implement a consumer-producer if we avoid callback pattern and only do the
group metadata initialization once, however the access pattern of consumer
rebalance state is scattered, which means we get both rebalance listener
and metadata getter. Jason's approach overloaded the initTransactions API,
which could be more confusing as it already has been today. Comparing the
two here, I'm inclined to Guozhang's approach as it is not conclusive to
say a new metadata getter class will confuse any user, with a sacrifice in
the cleanness of future implementation around consumer state. WDYT?

Boyang

On Wed, Aug 14, 2019 at 10:45 AM Guozhang Wang  wrote:

> My main concern is to require the overloaded `initTransactions` to be
> called repeatedly while the original `initTransactions` still called once
> throughout the life time, which is a bit confusing.
>
> Looking into the current POC PR, we actually only need the latest
> generation id when fetching offsets, so we can just make the GroupMetadata
> returned from the consumer a wrapper of the underlying values, and the
> getters of this object would always return the latest value.
> The values would be reset internally within the rebalances; and then the
> new `initTransactions` would still only be called once.
>
> Guozhang
>
>
> On Wed, Aug 14, 2019 at 9:53 AM Jason Gustafson 
> wrote:
>
> > Yeah, my reasoning is that the group metadata is only relevant to the
> > subscription API. So it makes sense to only expose it to the rebalance
> > listener.
> >
> > One option we could consider is bring back the `initTransactions`
> overload.
> > Then usage looks something like this:
> >
> > consumer.subscribe(topics, new RebalanceListener() {
> >   void onGroupJoined(GroupMetadata metadata) {
> > producer.initTransactions(metadata);
> >   }
> > }
> >
> > That seems pretty clean. What do you think?
> >
> > -Jason
> >
> > On Tue, Aug 13, 2019 at 6:07 PM Boyang Chen 
> > wrote:
> >
> > > Hey Guozhang,
> > >
> > > thanks for the suggestion. Could you elaborate more on why defining a
> > > direct consumer API would be easier? The benefit of reusing consumer
> > > rebalance listener is to consolidate the entry point of consumer
> internal
> > > states. Compared with letting consumer generate a deep-copy of metadata
> > > every time we call #sendOffsetsToTransactions, using a callback seems
> > > reducing unnecessary updates towards the metadata. WDYT?
> > >
> > > Boyang
> > >
> > > On Tue, Aug 13, 2019 at 2:14 PM Guozhang Wang 
> > wrote:
> > >
> > > > Hi Boyang, Jason,
> > > >
> > > > If we are going to expose the generation id / group.instance id etc
> > > anyways
> > > > I think its slightly better to just add a new API on KafkaConsumer
> > > > returning the ConsumerGroupMetadata (option 3) than passing it in on
> an
> > > > additional callback of ConsumerRebalanceListener.
> > > > It feels easier to leverage, than requiring users to pass in the
> > > listener.
> > > >
> > > > Guozhang
> > > >
> > > > On Mon, Aug 12, 2019 at 3:41 PM Boyang Chen <
> > reluctanthero...@gmail.com>
> > > > wrote:
> > > >
> > > > > Thanks Jason, the intuition behind defining a separate callback
> > > function
> > > > is
> > > > > that, with KIP-429 we no longer guarantee to call
> > > OnPartitionsAssigned()
> > > > or
> > > > > OnPartitionsRevoked() with each rebalance. Our requirement is to be
> > > > > up-to-date with group metadata such as generation information, so
> > > > callback
> > > > > like onGroupJoined() would make more sense as it should be invoked
> > > after
> > > > > every successful rebalance.
> > > > >
> > > > > Best,
> > > > > Boyang
> > > > >
> > > > > On Mon, Aug 12, 2019 at 2:02 PM Jason Gustafson <
> ja...@confluent.io>
> > > > > wrote:
> > > > >
> > > > > > Hey Boyang,
> > > > > >
> > > > > > I favor option 4 as well. It's a little more cumbersome than 3
> for
> > > this
> > > > > use
> > > > > > case, but it seems like a cleaner separation of concerns. The
> > > rebalance
> > > > > > listener is already concerned with events affecting the
> assignment
> > > > > > lifecycle and group membership. I think the only thing I'm
> > wondering
> > > is
> > > > > > whether it should be a separate callback as you've suggested, or
> if
> > > it
> > > > > > would make sense to overload `onPartitionsAssigned`. If it's
> > > separate,
> > > > > > maybe a name like `onGroupJoined` would be clearer?
> > > > > >
> > > > > > Thanks,
> > > > > > Jason
> > > > > >
> > > > > >
> > > > > >
> > > > > > On Thu, Aug 8, 2019 at 10:59 PM Boyang Chen <
> > > > reluctanthero...@gmail.com>
> > > > > > wrote:
> > > > > >
> > > > > > > Thank you Jason. We had some offline discussion on properly
> > keeping
> > > > > group
> > > > > > > metadata up to date, and here are some of our options
> > brainstormed:
> > > > > > > 1. Let the caller of `sendOffsetsToTransaction(offset,
> metadata)`
> > > > > >