[jira] [Created] (KAFKA-6291) Cannot close EmbeddedZookeeper on Windows

[Viliam Durina (JIRA)]

Viliam Durina created KAFKA-6291:


 Summary: Cannot close EmbeddedZookeeper on Windows
 Key: KAFKA-6291
 URL: https://issues.apache.org/jira/browse/KAFKA-6291
 Project: Kafka
  Issue Type: Bug
  Components: zkclient
Affects Versions: 1.0.0, 0.11.0.0
 Environment: Windows 10 (doesn't reproduce on Linux)
JDK 8
Reporter: Viliam Durina


We created {{EmbeddedZookeeper}} and {{ZkClient}} for various tests using this 
code:

{{
EmbeddedZookeeper zkServer = new EmbeddedZookeeper();
ZkClient zkClient = new ZkClient("127.0.0.1" + ':' + zkServer.port(), 
3, 3, ZKStringSerializer$.MODULE$);

zkClient.close();
zkServer.shutdown();
}}

This works fine on Linux, but on Windows, the {{zkServer.shutdown()}} call 
fails with this exception:

{{
[Thread-1] ERROR org.apache.kafka.test.TestUtils - Error deleting 
C:\Users\vilo\AppData\Local\Temp\kafka-5521621147877426083

java.nio.file.FileSystemException: 
C:\Users\vilo\AppData\Local\Temp\kafka-5521621147877426083\version-2\log.1: The 
process cannot access the file because it is being used by another process.

at 
sun.nio.fs.WindowsException.translateToIOException(WindowsException.java:86)
at 
sun.nio.fs.WindowsException.rethrowAsIOException(WindowsException.java:97)
at 
sun.nio.fs.WindowsException.rethrowAsIOException(WindowsException.java:102)
at 
sun.nio.fs.WindowsFileSystemProvider.implDelete(WindowsFileSystemProvider.java:269)
at 
sun.nio.fs.AbstractFileSystemProvider.delete(AbstractFileSystemProvider.java:103)
at java.nio.file.Files.delete(Files.java:1126)
at org.apache.kafka.common.utils.Utils$1.visitFile(Utils.java:630)
at org.apache.kafka.common.utils.Utils$1.visitFile(Utils.java:619)
at java.nio.file.Files.walkFileTree(Files.java:2670)
at java.nio.file.Files.walkFileTree(Files.java:2742)
at org.apache.kafka.common.utils.Utils.delete(Utils.java:619)
at org.apache.kafka.test.TestUtils$1.run(TestUtils.java:184)

java.nio.file.FileSystemException: 
C:\Users\vilo\AppData\Local\Temp\kafka-5521621147877426083\version-2\log.1: The 
process cannot access the file because it is being used by another process.


at 
sun.nio.fs.WindowsException.translateToIOException(WindowsException.java:86)
at 
sun.nio.fs.WindowsException.rethrowAsIOException(WindowsException.java:97)
at 
sun.nio.fs.WindowsException.rethrowAsIOException(WindowsException.java:102)
at 
sun.nio.fs.WindowsFileSystemProvider.implDelete(WindowsFileSystemProvider.java:269)
at 
sun.nio.fs.AbstractFileSystemProvider.delete(AbstractFileSystemProvider.java:103)
at java.nio.file.Files.delete(Files.java:1126)
at org.apache.kafka.common.utils.Utils$1.visitFile(Utils.java:630)
at org.apache.kafka.common.utils.Utils$1.visitFile(Utils.java:619)
at java.nio.file.Files.walkFileTree(Files.java:2670)
at java.nio.file.Files.walkFileTree(Files.java:2742)
at org.apache.kafka.common.utils.Utils.delete(Utils.java:619)
at kafka.zk.EmbeddedZookeeper.shutdown(EmbeddedZookeeper.scala:53)
at com.hazelcast.jet.KafkaTest.test(KafkaTest.java:32)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at 
org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at 
org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at 
org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at 
org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at 
com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
at 
com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(Idea

Re: [DISCUSS] KIP-227: Introduce Incremental FetchRequests to Increase Partition Scalability

[Becket Qin]

Hi Jan,

I agree that we probably don't want to make the protocol too complicated
just for exception cases.

The current FetchRequest contains an ordered list of partitions that may
rotate based on the priority. Therefore it is kind of difficult to do the
order matching. But you brought a good point about order, we may want to
migrate the rotation logic from the clients to the server. Not sure if this
will introduce some complexity to the broker. Intuitively it seems fine.
The logic would basically be similar to the draining logic in the
RecordAccumulator of the producer.

Thanks,

Jiangjie (Becket) Qin

On Thu, Nov 30, 2017 at 11:29 PM, Jan Filipiak 
wrote:

> Hi,
>
> this discussion is going a little bit far from what I intended this thread
> for.
> I can see all of this beeing related.
>
> To let you guys know what I am currently thinking is the following:
>
> I do think the handling of Id's and epoch is rather complicated. I think
> the complexity
> comes from aiming for to much.
>
> 1. Currently all the work is towards making fetchRequest
> completely empty. This brings all sorts of pain with regards to the broker
> actually needs
> to know what he send even though it tries to use sendfile as much as
> possible.
> 2. Currently all the work is towards also making empty fetch request
> across TCP sessions.
>
> In this thread I aimed to relax our goals with regards to point 2.
> Connection resets for us
> are really the exceptions and I would argue, trying to introduce
> complexity for sparing
> 1 full request on connection reset is not worth it. Therefore I argued to
> keep the Server
> side information with the Session instead somewhere global. Its not going
> to bring in the
> results.
>
> As the discussion unvields I also want to challenge our approach for point
> 1.
> I do not see a reason to introduce complexity (and
>  especially on the fetch answer path). Did we consider that from the
> client we just send the offsets
> we want to fetch and skip the topic partition description and just use the
> order to match the information
> on the broker side again? This would also reduce the fetch sizes a lot
> while skipping a ton of complexity.
>
> Hope these ideas are interesting
>
> best Jan
>
>
>
> On 01.12.2017 01:47, Becket Qin wrote:
>
>> Hi Colin,
>>
>> Thanks for updating the KIP. I have two comments:
>>
>> 1. The session epoch seems introducing some complexity. It would be good
>> if
>> we don't have to maintain the epoch.
>> 2. If all the partitions has data returned (even a few messages), the next
>> fetch would be equivalent to a full request. This means the clusters with
>> continuously small throughput may not save much from the incremental
>> fetch.
>>
>> I am wondering if we can avoid session epoch maintenance and address the
>> fetch efficiency in general with some modifications to the solution. Not
>> sure if the following would work, but just want to give my ideas.
>>
>> To solve 1, the basic idea is to let the leader return the partition data
>> with its expected client's position for each partition. If the client
>> disagree with the leader's expectation, a full FetchRequest is then sent
>> to
>> ask the leader to update the client's position.
>> To solve 2, when possible, we just let the leader to infer the clients
>> position instead of asking the clients to provide the position, so the
>> incremental fetch can be empty in most cases.
>>
>> More specifically, the protocol will have the following change.
>> 1. Add a new flag called FullFetch to the FetchRequest.
>> 1) A full FetchRequest is the same as the current FetchRequest with
>> FullFetch=true.
>> 2) An incremental FetchRequest is always empty with FullFetch=false.
>> 2. Add a new field called ExpectedPosition(INT64) to each partition data
>> in
>> the FetchResponse.
>>
>> The leader logic:
>> 1. The leader keeps a map from client-id (client-uuid) to the interested
>> partitions of that client. For each interested partition, the leader keeps
>> the client's position for that client.
>> 2. When the leader receives a full fetch request (FullFetch=true), the
>> leader
>>  1) replaces the interested partitions for the client id with the
>> partitions in that full fetch request.
>>  2) updates the client position with the offset specified in that full
>> fetch request.
>>  3) if the client is a follower, update the high watermark, etc.
>> 3. When the leader receives an incremental fetch request (typically
>> empty),
>> the leader returns the data from all the interested partitions (if any)
>> according to the position in the interested partitions map.
>> 4. In the FetchResponse, the leader will include an ExpectedFetchingOffset
>> that the leader thinks the client is fetching at. The value is the client
>> position of the partition in the interested partition map. This is just to
>> confirm with the client that the client position in the leader is correct.
>> 5. After sending back the FetchResponse, the leader upd

Re: [DISCUSS] KIP-227: Introduce Incremental FetchRequests to Increase Partition Scalability

[Jan Filipiak]

Hi,

this discussion is going a little bit far from what I intended this 
thread for.

I can see all of this beeing related.

To let you guys know what I am currently thinking is the following:

I do think the handling of Id's and epoch is rather complicated. I think 
the complexity

comes from aiming for to much.

1. Currently all the work is towards making fetchRequest
completely empty. This brings all sorts of pain with regards to the 
broker actually needs
to know what he send even though it tries to use sendfile as much as 
possible.
2. Currently all the work is towards also making empty fetch request 
across TCP sessions.


In this thread I aimed to relax our goals with regards to point 2. 
Connection resets for us
are really the exceptions and I would argue, trying to introduce 
complexity for sparing
1 full request on connection reset is not worth it. Therefore I argued 
to keep the Server
side information with the Session instead somewhere global. Its not 
going to bring in the

results.

As the discussion unvields I also want to challenge our approach for 
point 1.

I do not see a reason to introduce complexity (and
 especially on the fetch answer path). Did we consider that from the 
client we just send the offsets
we want to fetch and skip the topic partition description and just use 
the order to match the information
on the broker side again? This would also reduce the fetch sizes a lot 
while skipping a ton of complexity.


Hope these ideas are interesting

best Jan


On 01.12.2017 01:47, Becket Qin wrote:

Hi Colin,

Thanks for updating the KIP. I have two comments:

1. The session epoch seems introducing some complexity. It would be good if
we don't have to maintain the epoch.
2. If all the partitions has data returned (even a few messages), the next
fetch would be equivalent to a full request. This means the clusters with
continuously small throughput may not save much from the incremental fetch.

I am wondering if we can avoid session epoch maintenance and address the
fetch efficiency in general with some modifications to the solution. Not
sure if the following would work, but just want to give my ideas.

To solve 1, the basic idea is to let the leader return the partition data
with its expected client's position for each partition. If the client
disagree with the leader's expectation, a full FetchRequest is then sent to
ask the leader to update the client's position.
To solve 2, when possible, we just let the leader to infer the clients
position instead of asking the clients to provide the position, so the
incremental fetch can be empty in most cases.

More specifically, the protocol will have the following change.
1. Add a new flag called FullFetch to the FetchRequest.
1) A full FetchRequest is the same as the current FetchRequest with
FullFetch=true.
2) An incremental FetchRequest is always empty with FullFetch=false.
2. Add a new field called ExpectedPosition(INT64) to each partition data in
the FetchResponse.

The leader logic:
1. The leader keeps a map from client-id (client-uuid) to the interested
partitions of that client. For each interested partition, the leader keeps
the client's position for that client.
2. When the leader receives a full fetch request (FullFetch=true), the
leader
 1) replaces the interested partitions for the client id with the
partitions in that full fetch request.
 2) updates the client position with the offset specified in that full
fetch request.
 3) if the client is a follower, update the high watermark, etc.
3. When the leader receives an incremental fetch request (typically empty),
the leader returns the data from all the interested partitions (if any)
according to the position in the interested partitions map.
4. In the FetchResponse, the leader will include an ExpectedFetchingOffset
that the leader thinks the client is fetching at. The value is the client
position of the partition in the interested partition map. This is just to
confirm with the client that the client position in the leader is correct.
5. After sending back the FetchResponse, the leader updates the position of
the client's interested partitions. (There may be some overhead for the
leader to know of offsets, but I think the trick of returning at index
entry boundary or log end will work efficiently).
6. The leader will expire the client interested partitions if the client
hasn't fetch for some time. And if an incremental request is received when
the map does not contain the client info, an error will be returned to the
client to ask for a FullFetch.

The clients logic:
1. Start with sending a full FetchRequest, including partitions and offsets.
2. When get a response, check the ExpectedOffsets in the fetch response and
see if that matches the current log end.
 1) If the ExpectedFetchOffset matches the current log end, the next
fetch request will be an incremental fetch request.
 2) if the ExpectedFetchOffset does not match the current log end, the
next fe

[jira] [Created] (KAFKA-6290) Kafka Connect fails when using cast transformation

[Sudhir Pradhan (JIRA)]

Sudhir Pradhan created KAFKA-6290:
-

 Summary: Kafka Connect fails when using cast transformation
 Key: KAFKA-6290
 URL: https://issues.apache.org/jira/browse/KAFKA-6290
 Project: Kafka
  Issue Type: Task
  Components: KafkaConnect
Affects Versions: 0.11.0.0
Reporter: Sudhir Pradhan


I am facing same issue when consuming from KAFKA to HDFS with CAST TRANSFORMS. 
Any pointer please.

My Connector :
*
```
{
 "name": "hdfs-sink-avro-cast-test-stndln",
 "config": {
  "key.converter": "io.confluent.connect.avro.AvroConverter",
  "key.converter.schema.registry.url": "http://localhost:8081";,
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "http://localhost:8081";,
  "key.converter.schemas.enable": "true",
  "value.converter.schemas.enable": "true",
  "internal.key.converter": "org.apache.kafka.connect.json.JsonConverter",
  "internal.value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "internal.key.converter.schemas.enable": "false",
  "internal.value.converter.schemas.enable": "false",
  "offset.storage.file.filename": 
"/tmp/connect.offsets.avroHdfsConsumer.casttest.stndln",
  "offset.flush.interval.ms": "500",
  "parse.key": "true",
  "connector.class": "io.confluent.connect.hdfs.HdfsSinkConnector",
  "hadoop.home": "/usr/lib/hadoop",
  "hdfs.url": "hdfs://ip-127-34-56-789.us-east-1.compute.interna:8020",
  "topics": "avro_raw_KFK_SRP_USER_TEST_V,avro_raw_KFK_SRP_PG_HITS_TEST_V",
  "tasks.max": "1",
  "topics.dir": "/home/hadoop/kafka/data/streams/in/raw/casttest1",
  "logs.dir": "/home/hadoop/kafka/wal/streams/in/raw/casttest1",
  "hive.integration": "true",
  "hive.metastore.uris": 
"thrift://ip-127-34-56-789.us-east-1.compute.internal:9083",
  "schema.compatibility": "BACKWARD",
  "flush.size": "1",
  "rotate.interval.ms": "1000",
  "mode": "timestamp",
  "transforms": "Cast",
  "transforms.Cast.type": "org.apache.kafka.connect.transforms.Cast$Value",
  "transforms.Cast.spec": "residuals:float64,comp:float64"
 }
}
```

Exception :
*
```
[2017-11-16 01:14:39,719] ERROR Task hdfs-sink-avro-cast-test-stndln-0 threw an 
uncaught and unrecoverable exception 
(org.apache.kafka.connect.runtime.WorkerTask:148)
org.apache.kafka.connect.errors.DataException: Invalid Java object for schema 
type INT64: class java.util.Date for field: "null"
at 
org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:239)
at 
org.apache.kafka.connect.data.ConnectSchema.validateValue(ConnectSchema.java:209)
at org.apache.kafka.connect.data.Struct.put(Struct.java:214)
at 
org.apache.kafka.connect.transforms.Cast.applyWithSchema(Cast.java:152)
at org.apache.kafka.connect.transforms.Cast.apply(Cast.java:108)
at 
org.apache.kafka.connect.runtime.TransformationChain.apply(TransformationChain.java:38)
at 
org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:414)
at 
org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:250)
at 
org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:180)
at 
org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:148)
at 
org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:146)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:190)
at 
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at 
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at 
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
[2017-11-16 01:14:39,719] ERROR Task is being killed and will not recover until 
manually restarted (org.apache.kafka.connect.runtime.WorkerTask:149)
[2017-11-16 01:14:39,719] INFO Shutting down Hive executor service. 
(io.confluent.connect.hdfs.DataWriter:309)
```




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #4280: KAFKA-6289: NetworkClient should not expose failed...

[hachikuji]

GitHub user hachikuji opened a pull request:

https://github.com/apache/kafka/pull/4280

KAFKA-6289: NetworkClient should not expose failed internal ApiVersions 
requests

The NetworkClient internally ApiVersion requests to each broker following 
connection establishment. If this request happens to fail (perhaps due to an 
incompatible broker), the NetworkClient includes the response in the result of 
poll(). Applications will generally not be expecting this response which may 
lead to failed assertions (or in the case of AdminClient, an obscure log 
message).

I've added test cases which await the ApiVersion request sent by 
NetworkClient to be in-flight, and then disconnect the connection and verify 
that the response is not included from poll().

### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation 
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)


You can merge this pull request into a Git repository by running:

$ git pull https://github.com/hachikuji/kafka KAFKA-6289

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/4280.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #4280


commit 3c7420b2ab51f8d918037681ed9862f0d2676ba7
Author: Jason Gustafson 
Date:   2017-12-01T05:03:54Z

KAFKA-6289: NetworkClient should not expose failed internal ApiVersion 
requests




---

[jira] [Created] (KAFKA-6289) NetworkClient should not return internal failed api version responses from poll

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-6289:
--

 Summary: NetworkClient should not return internal failed api 
version responses from poll
 Key: KAFKA-6289
 URL: https://issues.apache.org/jira/browse/KAFKA-6289
 Project: Kafka
  Issue Type: Bug
Reporter: Jason Gustafson
Assignee: Jason Gustafson


In the AdminClient, if the initial ApiVersion request sent to the broker fails, 
we see the following obscure message:
{code}
[2017-11-30 17:18:48,677] ERROR Internal server error on -2: server returned 
information about unknown correlation ID 0.  requestHeader = 
{api_key=18,api_version=1,correlation_id=0,client_id=adminclient-3} 
(org.apache.kafka.clients.admin.KafkaAdminClient)
{code}
What's happening is that the response to the internal ApiVersion request which 
is received in NetworkClient is mistakenly being sent to the upper layer (the 
admin client in this case). The admin wasn't expecting it, so we see this 
message. Instead, the request should be handled internally in NetworkClient.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Created] (KAFKA-6288) Broken symlink interrupts scanning the plugin path

[Yeva Byzek (JIRA)]

Yeva Byzek created KAFKA-6288:
-

 Summary: Broken symlink interrupts scanning the plugin path
 Key: KAFKA-6288
 URL: https://issues.apache.org/jira/browse/KAFKA-6288
 Project: Kafka
  Issue Type: Bug
  Components: KafkaConnect
Affects Versions: 1.0.0
Reporter: Yeva Byzek


KAFKA-6087 introduced support for scanning relative symlinks in the plugin 
path. However, if a relative symlink points to a target that doesn't exist, 
then scanning the plugin path is interrupted.

Desired behavior is that the symlink is skipped and scanning the plugin path 
continues.

Example of error message:

{noformat}
[2017-11-30 20:19:26,226] ERROR Could not get listing for plugin path: 
/usr/share/java. Ignoring. 
(org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:170)
java.nio.file.NoSuchFileException: /usr/share/java/name.jar
at 
sun.nio.fs.UnixException.translateToIOException(UnixException.java:86)
at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:102)
at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:107)
at sun.nio.fs.UnixPath.toRealPath(UnixPath.java:837)
at 
org.apache.kafka.connect.runtime.isolation.PluginUtils.pluginUrls(PluginUtils.java:241)
at 
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.registerPlugin(DelegatingClassLoader.java:181)
at 
org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader.initLoaders(DelegatingClassLoader.java:153)
at 
org.apache.kafka.connect.runtime.isolation.Plugins.(Plugins.java:47)
at 
org.apache.kafka.connect.cli.ConnectDistributed.main(ConnectDistributed.java:70)
{noformat}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Re: [DISCUSS] KIP-227: Introduce Incremental FetchRequests to Increase Partition Scalability

[Becket Qin]

Hi Colin,

Thanks for updating the KIP. I have two comments:

1. The session epoch seems introducing some complexity. It would be good if
we don't have to maintain the epoch.
2. If all the partitions has data returned (even a few messages), the next
fetch would be equivalent to a full request. This means the clusters with
continuously small throughput may not save much from the incremental fetch.

I am wondering if we can avoid session epoch maintenance and address the
fetch efficiency in general with some modifications to the solution. Not
sure if the following would work, but just want to give my ideas.

To solve 1, the basic idea is to let the leader return the partition data
with its expected client's position for each partition. If the client
disagree with the leader's expectation, a full FetchRequest is then sent to
ask the leader to update the client's position.
To solve 2, when possible, we just let the leader to infer the clients
position instead of asking the clients to provide the position, so the
incremental fetch can be empty in most cases.

More specifically, the protocol will have the following change.
1. Add a new flag called FullFetch to the FetchRequest.
   1) A full FetchRequest is the same as the current FetchRequest with
FullFetch=true.
   2) An incremental FetchRequest is always empty with FullFetch=false.
2. Add a new field called ExpectedPosition(INT64) to each partition data in
the FetchResponse.

The leader logic:
1. The leader keeps a map from client-id (client-uuid) to the interested
partitions of that client. For each interested partition, the leader keeps
the client's position for that client.
2. When the leader receives a full fetch request (FullFetch=true), the
leader
1) replaces the interested partitions for the client id with the
partitions in that full fetch request.
2) updates the client position with the offset specified in that full
fetch request.
3) if the client is a follower, update the high watermark, etc.
3. When the leader receives an incremental fetch request (typically empty),
the leader returns the data from all the interested partitions (if any)
according to the position in the interested partitions map.
4. In the FetchResponse, the leader will include an ExpectedFetchingOffset
that the leader thinks the client is fetching at. The value is the client
position of the partition in the interested partition map. This is just to
confirm with the client that the client position in the leader is correct.
5. After sending back the FetchResponse, the leader updates the position of
the client's interested partitions. (There may be some overhead for the
leader to know of offsets, but I think the trick of returning at index
entry boundary or log end will work efficiently).
6. The leader will expire the client interested partitions if the client
hasn't fetch for some time. And if an incremental request is received when
the map does not contain the client info, an error will be returned to the
client to ask for a FullFetch.

The clients logic:
1. Start with sending a full FetchRequest, including partitions and offsets.
2. When get a response, check the ExpectedOffsets in the fetch response and
see if that matches the current log end.
1) If the ExpectedFetchOffset matches the current log end, the next
fetch request will be an incremental fetch request.
2) if the ExpectedFetchOffset does not match the current log end, the
next fetch request will be a full FetchRequest.
3. Whenever the partition offset is actively changed (e.g. consumer.seek(),
follower log truncation, etc), a full FetchRequest is sent.
4. Whenever the interested partition set changes (e.g.
consumer.subscribe()/assign() is called, replica reassignment happens), a
full FetchRequest is sent.
5. Whenever the client needs to retry a fetch, a FullFetch is sent.

The benefits of this approach are:
1. Regardless of the traffic pattern in the cluster, In most cases the
fetch request will be empty.
2. No need to maintain session epochs.

What do you think?

Thanks,

Jiangjie (Becket) Qin


On Thu, Nov 30, 2017 at 9:37 AM, Colin McCabe  wrote:

> On Wed, Nov 29, 2017, at 18:59, Dong Lin wrote:
> > Hey Colin,
> >
> > Thanks much for the update. I have a few questions below:
> >
> > 1. I am not very sure that we need Fetch Session Epoch. It seems that
> > Fetch
> > Session Epoch is only needed to help leader distinguish between "a full
> > fetch request" and "a full fetch request and request a new incremental
> > fetch session". Alternatively, follower can also indicate "a full fetch
> > request and request a new incremental fetch session" by setting Fetch
> > Session ID to -1 without using Fetch Session Epoch. Does this make sense?
>
> Hi Dong,
>
> The fetch session epoch is very important for ensuring correctness.  It
> prevents corrupted or incomplete fetch data due to network reordering or
> loss.
>
> For example, consider a scenario where the follower sends a fetch
> request to the leader.  The l

Build failed in Jenkins: kafka-trunk-jdk9 #231

[Apache Jenkins Server]

See 


Changes:

[me] KAFKA-6284: Fixed system test for Connect REST API

--
[...truncated 1.44 MB...]
kafka.zk.ZKPathTest > testMakeSurePersistsPathExists PASSED

kafka.zk.KafkaZkClientTest > testSetGetAndDeletePartitionReassignment STARTED

kafka.zk.KafkaZkClientTest > testSetGetAndDeletePartitionReassignment PASSED

kafka.zk.KafkaZkClientTest > testGetDataAndVersion STARTED

kafka.zk.KafkaZkClientTest > testGetDataAndVersion PASSED

kafka.zk.KafkaZkClientTest > testGetChildren STARTED

kafka.zk.KafkaZkClientTest > testGetChildren PASSED

kafka.zk.KafkaZkClientTest > testSetAndGetConsumerOffset STARTED

kafka.zk.KafkaZkClientTest > testSetAndGetConsumerOffset PASSED

kafka.zk.KafkaZkClientTest > testEntityConfigManagementMethods STARTED

kafka.zk.KafkaZkClientTest > testEntityConfigManagementMethods PASSED

kafka.zk.KafkaZkClientTest > testCreateRecursive STARTED

kafka.zk.KafkaZkClientTest > testCreateRecursive PASSED

kafka.zk.KafkaZkClientTest > testGetConsumerOffsetNoData STARTED

kafka.zk.KafkaZkClientTest > testGetConsumerOffsetNoData PASSED

kafka.zk.KafkaZkClientTest > testDeleteTopicPathMethods STARTED

kafka.zk.KafkaZkClientTest > testDeleteTopicPathMethods PASSED

kafka.zk.KafkaZkClientTest > testAclManagementMethods STARTED

kafka.zk.KafkaZkClientTest > testAclManagementMethods PASSED

kafka.zk.KafkaZkClientTest > testGetDataAndStat STARTED

kafka.zk.KafkaZkClientTest > testGetDataAndStat PASSED

kafka.zk.KafkaZkClientTest > testConditionalUpdatePath STARTED

kafka.zk.KafkaZkClientTest > testConditionalUpdatePath PASSED

kafka.zk.KafkaZkClientTest > testTopicAssignmentMethods STARTED

kafka.zk.KafkaZkClientTest > testTopicAssignmentMethods PASSED

kafka.zk.KafkaZkClientTest > testDeleteRecursive STARTED

kafka.zk.KafkaZkClientTest > testDeleteRecursive PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytesWithCompression 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytesWithCompression 
PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > 
testIteratorIsConsistentWithCompression STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > 
testIteratorIsConsistentWithCompression PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testIteratorIsConsistent 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testIteratorIsConsistent PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testEqualsWithCompression 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testEqualsWithCompression 
PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testWrittenEqualsRead STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testWrittenEqualsRead PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testEquals STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testEquals PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytes STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytes PASSED

kafka.javaapi.consumer.ZookeeperConsumerConnectorTest > testBasic STARTED

kafka.javaapi.consumer.ZookeeperConsumerConnectorTest > testBasic PASSED

kafka.metrics.MetricsTest > testMetricsReporterAfterDeletingTopic STARTED

kafka.metrics.MetricsTest > testMetricsReporterAfterDeletingTopic PASSED

kafka.metrics.MetricsTest > 
testBrokerTopicMetricsUnregisteredAfterDeletingTopic STARTED

kafka.metrics.MetricsTest > 
testBrokerTopicMetricsUnregisteredAfterDeletingTopic PASSED

kafka.metrics.MetricsTest > testClusterIdMetric STARTED

kafka.metrics.MetricsTest > testClusterIdMetric PASSED

kafka.metrics.MetricsTest > testControllerMetrics STARTED

kafka.metrics.MetricsTest > testControllerMetrics PASSED

kafka.metrics.MetricsTest > testWindowsStyleTagNames STARTED

kafka.metrics.MetricsTest > testWindowsStyleTagNames PASSED

kafka.metrics.MetricsTest > testMetricsLeak STARTED

kafka.metrics.MetricsTest > testMetricsLeak PASSED

kafka.metrics.MetricsTest > testBrokerTopicMetricsBytesInOut STARTED

kafka.metrics.MetricsTest > testBrokerTopicMetricsBytesInOut PASSED

kafka.metrics.KafkaTimerTest > testKafkaTimer STARTED

kafka.metrics.KafkaTimerTest > testKafkaTimer PASSED

kafka.security.auth.ResourceTypeTest > testJavaConversions STARTED

kafka.security.auth.ResourceTypeTest > testJavaConversions PASSED

kafka.security.auth.ResourceTypeTest > testFromString STARTED

kafka.security.auth.ResourceTypeTest > testFromString PASSED

kafka.security.auth.OperationTest > testJavaConversions STARTED

kafka.security.auth.OperationTest > testJavaConversions PASSED

kafka.security.auth.ZkAuthorizationTest > testIsZkSecurityEnabled STARTED

kafka.security.auth.ZkAuthorizationTest > testIsZkSecurityEnabled PASSED

kafka.security.auth.ZkAuthorizationTest > testZkUtils STARTED

kafka.security.auth.ZkAuthorizationTest > testZkUtils PASSED

kafka.security.auth.ZkAuthorizationTest > testZkAntiMigration STARTED

[GitHub] kafka pull request #4279: KAFKA-6284: Fixed system test for Connect REST API

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/4279


---

[jira] [Resolved] (KAFKA-6284) System Test failed: ConnectRestApiTest

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava resolved KAFKA-6284.
--
Resolution: Fixed

Issue resolved by pull request 4279
[https://github.com/apache/kafka/pull/4279]

> System Test failed: ConnectRestApiTest 
> ---
>
> Key: KAFKA-6284
> URL: https://issues.apache.org/jira/browse/KAFKA-6284
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Affects Versions: 1.1.0
>Reporter: Mikkin Patel
>Assignee: Mikkin Patel
> Fix For: 1.1.0
>
>
> KAFKA-3073 introduced topic regex support for Connect sinks. The 
> ConnectRestApiTest failed to verifiy configdef with expected response. 
> {noformat}
> Traceback (most recent call last):
>   File 
> "/home/jenkins/workspace/system-test-kafka-ci_trunk-STZ4BWS25LWQAJN6JJZHKK6GP4YF7HCVYRFRFXB2KMAQ4PC5H4ZA/kafka/venv/local/lib/python2.7/site-packages/ducktape-0.7.1-py2.7.egg/ducktape/tests/runner_client.py",
>  line 132, in run
> data = self.run_test()
>   File 
> "/home/jenkins/workspace/system-test-kafka-ci_trunk-STZ4BWS25LWQAJN6JJZHKK6GP4YF7HCVYRFRFXB2KMAQ4PC5H4ZA/kafka/venv/local/lib/python2.7/site-packages/ducktape-0.7.1-py2.7.egg/ducktape/tests/runner_client.py",
>  line 185, in run_test
> return self.test_context.function(self.test)
>   File 
> "/home/jenkins/workspace/system-test-kafka-ci_trunk-STZ4BWS25LWQAJN6JJZHKK6GP4YF7HCVYRFRFXB2KMAQ4PC5H4ZA/kafka/tests/kafkatest/tests/connect/connect_rest_test.py",
>  line 92, in test_rest_api
> self.verify_config(self.FILE_SINK_CONNECTOR, self.FILE_SINK_CONFIGS, 
> configs)
>   File 
> "/home/jenkins/workspace/system-test-kafka-ci_trunk-STZ4BWS25LWQAJN6JJZHKK6GP4YF7HCVYRFRFXB2KMAQ4PC5H4ZA/kafka/tests/kafkatest/tests/connect/connect_rest_test.py",
>  line 200, in verify_config
> assert config_def == set(config_names)
> AssertionError
> {noformat}



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[jira] [Created] (KAFKA-6287) Inconsistent protocol type for empty consumer groups

[Ryan Leslie (JIRA)]

Ryan Leslie created KAFKA-6287:
--

 Summary: Inconsistent protocol type for empty consumer groups
 Key: KAFKA-6287
 URL: https://issues.apache.org/jira/browse/KAFKA-6287
 Project: Kafka
  Issue Type: Bug
  Components: consumer
Affects Versions: 1.0.0
Reporter: Ryan Leslie
Priority: Minor


When a consumer is created for a new group, the group metadata's protocol type 
is set to 'consumer' and this is stored both in __consumer_offsets as well as 
in the coordinator's local cache.

If the consumer leaves the group and the group becomes empty, ListGroups 
requests will continue to show the group as type 'consumer', and as such 
kafka-consumer-groups.sh will show it via --list.

However, if the coordinator (broker) node is killed and a new coordinator is 
elected, when the GroupMetadataManager loads the group from __consumer_offsets 
into its cache, it will not set the protocolType if there are no active 
consumers. As a result, the group's protocolType will now become the empty 
string (UNKNOWN_PROTOCOL_TYPE), and kafka-consumer-groups.sh will no longer 
show the group.

Ideally bouncing a broker should not result in the group's protocol changing. 
protocolType can be set in GroupMetadataManager.readGroupMessageValue() to 
always reflect what's present in the persistent metadata (__consumer_offsets) 
regardless if there are active members.


In general, things can get confusing when distinguishing between 'consumer' and 
non-consumer groups. For example, DescribeGroups and OffsetFetchRequest does 
not filter on protocol type, so it's possible for kafka-consumer-groups.sh to 
describe groups (--describe) without actually being able to list them. In the 
protocol guide, OffsetFetchRequest / OffsetCommitRequest have their parameters 
listed as 'ConsumerGroup', but in reality these can be used for groups of 
unknown type as well. For instance, a consumer group can be copied by finding a 
coordinator (GroupCoordinatorRequest / FindCoordinatorRequest) and sending an 
OffsetCommitRequest. The group will be auto-created with an empty protocol. 
Although this is arguably correct, the group will now exist but not be  a 
proper 'consumer' group until later when there is a JoinGroupRequest. Again, 
this can be confusing as far as categorization / visibility of the group is 
concerned. A group can instead be copied more directly by creating a consumer 
and calling commitSync (as kafka-consumer-groups.sh), but this does involve 
extra connections / requests and so is a little slower when trying to keep a 
large number of groups in sync in real-time across clusters.

If we want to make it easier to keep consumer groups consistent, options 
include allowing groups to be explicitly created with a protocol type instead 
of only lazily, or for groups created outside of JoinGroupRequest the 
coordinator can gain a broker config to set a default protocol type for groups. 
This would default to 'consumer'. This sort of setting might be confusing for 
users though, since implicitly created groups is certainly not the norm.



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #4279: KAFKA-6284: Fixed system test for Connect REST API

[mikkin]

GitHub user mikkin opened a pull request:

https://github.com/apache/kafka/pull/4279

KAFKA-6284: Fixed system test for Connect REST API

`topics.regex` was added in KAFKA-3073. This change fixes the test that 
invokes `/validate` to ensure that all the configdefs are returned as expected.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/mikkin/kafka KAFKA-6284

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/4279.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #4279


commit 426dfa9b888acbe6f8c5f90b1c24dad62f92ded7
Author: Mikkin 
Date:   2017-11-30T17:37:56Z

KAFKA-6284: Fixed system test for Connect REST API




---

Re: [DISCUSS] KIP-227: Introduce Incremental FetchRequests to Increase Partition Scalability

[Colin McCabe]

On Wed, Nov 29, 2017, at 18:59, Dong Lin wrote:
> Hey Colin,
> 
> Thanks much for the update. I have a few questions below:
> 
> 1. I am not very sure that we need Fetch Session Epoch. It seems that
> Fetch
> Session Epoch is only needed to help leader distinguish between "a full
> fetch request" and "a full fetch request and request a new incremental
> fetch session". Alternatively, follower can also indicate "a full fetch
> request and request a new incremental fetch session" by setting Fetch
> Session ID to -1 without using Fetch Session Epoch. Does this make sense?

Hi Dong,

The fetch session epoch is very important for ensuring correctness.  It
prevents corrupted or incomplete fetch data due to network reordering or
loss.

For example, consider a scenario where the follower sends a fetch
request to the leader.  The leader responds, but the response is lost
because of network problems which affected the TCP session.  In that
case, the follower must establish a new TCP session and re-send the
incremental fetch request.  But the leader does not know that the
follower didn't receive the previous incremental fetch response.  It is
only the incremental fetch epoch which lets the leader know that it
needs to resend that data, and not data which comes afterwards.

You could construct similar scenarios with message reordering,
duplication, etc.  Basically, this is a stateful protocol on an
unreliable network, and you need to know whether the follower got the
previous data you sent before you move on.  And you need to handle
issues like duplicated or delayed requests.  These issues do not affect
the full fetch request, because it is not stateful-- any full fetch
request can be understood and properly responded to in isolation.

> 
> 2. It is said that Incremental FetchRequest will include partitions whose
> fetch offset or maximum number of fetch bytes has been changed. If
> follower's logStartOffet of a partition has changed, should this
> partition also be included in the next FetchRequest to the leader? Otherwise, 
> it
> may affect the handling of DeleteRecordsRequest because leader may not know
> the corresponding data has been deleted on the follower.

Yeah, the follower should include the partition if the logStartOffset
has changed.  That should be spelled out on the KIP.  Fixed.

> 
> 3. In the section "Per-Partition Data", a partition is not considered
> dirty if its log start offset has changed. Later in the section "FetchRequest
> Changes", it is said that incremental fetch responses will include a
> partition if its logStartOffset has changed. It seems inconsistent. Can
> you update the KIP to clarify it?
> 

In the "Per-Partition Data" section, it does say that logStartOffset
changes make a partition dirty, though, right?  The first bullet point
is:

> * The LogCleaner deletes messages, and this changes the log start offset of 
> the partition on the leader., or

> 4. In "Fetch Session Caching" section, it is said that each broker has a
> limited number of slots. How is this number determined? Does this require
> a new broker config for this number?

Good point.  I added two broker configuration parameters to control this
number.

> What is the error code if broker does
> not have new log for the incoming FetchRequest?

Hmm, is there a typo in this question?  Maybe you meant to ask what
happens if there is no new cache slot for the incoming FetchRequest? 
That's not an error-- the incremental fetch session ID just gets set to
0, indicating no incremental fetch session was created.

> 
> 5. Can you clarify what happens if follower adds a partition to the
> ReplicaFetcherThread after receiving LeaderAndIsrRequest? Does leader
> needs to generate a new session for this ReplicaFetcherThread or does it 
> re-use
> the existing session?  If it uses a new session, is the old session
> actively deleted from the slot?

The basic idea is that you can't make changes, except by sending a full
fetch request.  However, perhaps we can allow the client to re-use its
existing session ID.  If the client sets sessionId = id, epoch = 0, it
could re-initialize the session.

> 
> 
> BTW, I think it may be useful if the KIP can include the example workflow
> of how this feature will be used in case of partition change and so on.

Yeah, that might help.

best,
Colin

> 
> Thanks,
> Dong
> 
> 
> On Wed, Nov 29, 2017 at 12:13 PM, Colin McCabe 
> wrote:
> 
> > I updated the KIP with the ideas we've been discussing.
> >
> > best,
> > Colin
> >
> > On Tue, Nov 28, 2017, at 08:38, Colin McCabe wrote:
> > > On Mon, Nov 27, 2017, at 22:30, Jan Filipiak wrote:
> > > > Hi Colin, thank you  for this KIP, it can become a really useful thing.
> > > >
> > > > I just scanned through the discussion so far and wanted to start a
> > > > thread to make as decision about keeping the
> > > > cache with the Connection / Session or having some sort of UUID indN
> > exed
> > > > global Map.
> > > >
> > > > Sorry if that has been settled already

Doc says when doing re-balance, sort by leader then partition, but the code seems sort only on partition

[李响]

Dear Kafka community,

In the doc -> https://kafka.apache.org/documentation/#distributionimpl
4. sort Pt (so partitions on the same broker are clustered together)
and
During rebalancing, we try to assign partitions to consumers in such a way
that reduces the number of broker nodes each consumer has to connect to

If I get it correctly, it means when sorting the partitions, firstly it is
sorted by leader. Then among those partitions which have the same leader,
then it is sorted by partition id in numeric order.

While the code of RangeAssignor of
https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/consumer/PartitionAssignor.scala
seems tell that it is sorted only on partition, not on leader.

My trial also demonstrated that the code is correct.

Is the doc out of date? Or the doc follows some previous versions of Kafka?
Or my understanding on code has something wrong?

Please kindly advise, thanks!!


-- 

   李响 Xiang Li

邮件 e-mail  ：wate...@gmail.com

Build failed in Jenkins: kafka-trunk-jdk9 #230

[Apache Jenkins Server]

See 

--
[...truncated 1.45 MB...]
kafka.zk.ZKPathTest > testMakeSurePersistsPathExists PASSED

kafka.zk.KafkaZkClientTest > testSetGetAndDeletePartitionReassignment STARTED

kafka.zk.KafkaZkClientTest > testSetGetAndDeletePartitionReassignment PASSED

kafka.zk.KafkaZkClientTest > testGetDataAndVersion STARTED

kafka.zk.KafkaZkClientTest > testGetDataAndVersion PASSED

kafka.zk.KafkaZkClientTest > testGetChildren STARTED

kafka.zk.KafkaZkClientTest > testGetChildren PASSED

kafka.zk.KafkaZkClientTest > testSetAndGetConsumerOffset STARTED

kafka.zk.KafkaZkClientTest > testSetAndGetConsumerOffset PASSED

kafka.zk.KafkaZkClientTest > testEntityConfigManagementMethods STARTED

kafka.zk.KafkaZkClientTest > testEntityConfigManagementMethods PASSED

kafka.zk.KafkaZkClientTest > testCreateRecursive STARTED

kafka.zk.KafkaZkClientTest > testCreateRecursive PASSED

kafka.zk.KafkaZkClientTest > testGetConsumerOffsetNoData STARTED

kafka.zk.KafkaZkClientTest > testGetConsumerOffsetNoData PASSED

kafka.zk.KafkaZkClientTest > testDeleteTopicPathMethods STARTED

kafka.zk.KafkaZkClientTest > testDeleteTopicPathMethods PASSED

kafka.zk.KafkaZkClientTest > testAclManagementMethods STARTED

kafka.zk.KafkaZkClientTest > testAclManagementMethods PASSED

kafka.zk.KafkaZkClientTest > testGetDataAndStat STARTED

kafka.zk.KafkaZkClientTest > testGetDataAndStat PASSED

kafka.zk.KafkaZkClientTest > testConditionalUpdatePath STARTED

kafka.zk.KafkaZkClientTest > testConditionalUpdatePath PASSED

kafka.zk.KafkaZkClientTest > testTopicAssignmentMethods STARTED

kafka.zk.KafkaZkClientTest > testTopicAssignmentMethods PASSED

kafka.zk.KafkaZkClientTest > testDeleteRecursive STARTED

kafka.zk.KafkaZkClientTest > testDeleteRecursive PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytesWithCompression 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytesWithCompression 
PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > 
testIteratorIsConsistentWithCompression STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > 
testIteratorIsConsistentWithCompression PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testIteratorIsConsistent 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testIteratorIsConsistent PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testEqualsWithCompression 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testEqualsWithCompression 
PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testWrittenEqualsRead STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testWrittenEqualsRead PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testEquals STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testEquals PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytes STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytes PASSED

kafka.javaapi.consumer.ZookeeperConsumerConnectorTest > testBasic STARTED

kafka.javaapi.consumer.ZookeeperConsumerConnectorTest > testBasic PASSED

kafka.metrics.MetricsTest > testMetricsReporterAfterDeletingTopic STARTED

kafka.metrics.MetricsTest > testMetricsReporterAfterDeletingTopic PASSED

kafka.metrics.MetricsTest > 
testBrokerTopicMetricsUnregisteredAfterDeletingTopic STARTED

kafka.metrics.MetricsTest > 
testBrokerTopicMetricsUnregisteredAfterDeletingTopic PASSED

kafka.metrics.MetricsTest > testClusterIdMetric STARTED

kafka.metrics.MetricsTest > testClusterIdMetric PASSED

kafka.metrics.MetricsTest > testControllerMetrics STARTED

kafka.metrics.MetricsTest > testControllerMetrics PASSED

kafka.metrics.MetricsTest > testWindowsStyleTagNames STARTED

kafka.metrics.MetricsTest > testWindowsStyleTagNames PASSED

kafka.metrics.MetricsTest > testMetricsLeak STARTED

kafka.metrics.MetricsTest > testMetricsLeak PASSED

kafka.metrics.MetricsTest > testBrokerTopicMetricsBytesInOut STARTED

kafka.metrics.MetricsTest > testBrokerTopicMetricsBytesInOut PASSED

kafka.metrics.KafkaTimerTest > testKafkaTimer STARTED

kafka.metrics.KafkaTimerTest > testKafkaTimer PASSED

kafka.security.auth.ResourceTypeTest > testJavaConversions STARTED

kafka.security.auth.ResourceTypeTest > testJavaConversions PASSED

kafka.security.auth.ResourceTypeTest > testFromString STARTED

kafka.security.auth.ResourceTypeTest > testFromString PASSED

kafka.security.auth.OperationTest > testJavaConversions STARTED

kafka.security.auth.OperationTest > testJavaConversions PASSED

kafka.security.auth.ZkAuthorizationTest > testIsZkSecurityEnabled STARTED

kafka.security.auth.ZkAuthorizationTest > testIsZkSecurityEnabled PASSED

kafka.security.auth.ZkAuthorizationTest > testZkUtils STARTED

kafka.security.auth.ZkAuthorizationTest > testZkUtils PASSED

kafka.security.auth.ZkAuthorizationTest > testZkAntiMigration STARTED

kafka.security.auth.ZkAuthorizationTest > testZkAntiMigration PASSED

kafka.secu

Build failed in Jenkins: kafka-trunk-jdk9 #229

[Apache Jenkins Server]

See 


Changes:

[ismael] MINOR: Shutdown ControllerEventThread via event instead of interruption

--
[...truncated 1.45 MB...]
kafka.zk.ZKPathTest > testMakeSurePersistsPathExists PASSED

kafka.zk.KafkaZkClientTest > testSetGetAndDeletePartitionReassignment STARTED

kafka.zk.KafkaZkClientTest > testSetGetAndDeletePartitionReassignment PASSED

kafka.zk.KafkaZkClientTest > testGetDataAndVersion STARTED

kafka.zk.KafkaZkClientTest > testGetDataAndVersion PASSED

kafka.zk.KafkaZkClientTest > testGetChildren STARTED

kafka.zk.KafkaZkClientTest > testGetChildren PASSED

kafka.zk.KafkaZkClientTest > testSetAndGetConsumerOffset STARTED

kafka.zk.KafkaZkClientTest > testSetAndGetConsumerOffset PASSED

kafka.zk.KafkaZkClientTest > testEntityConfigManagementMethods STARTED

kafka.zk.KafkaZkClientTest > testEntityConfigManagementMethods PASSED

kafka.zk.KafkaZkClientTest > testCreateRecursive STARTED

kafka.zk.KafkaZkClientTest > testCreateRecursive PASSED

kafka.zk.KafkaZkClientTest > testGetConsumerOffsetNoData STARTED

kafka.zk.KafkaZkClientTest > testGetConsumerOffsetNoData PASSED

kafka.zk.KafkaZkClientTest > testDeleteTopicPathMethods STARTED

kafka.zk.KafkaZkClientTest > testDeleteTopicPathMethods PASSED

kafka.zk.KafkaZkClientTest > testAclManagementMethods STARTED

kafka.zk.KafkaZkClientTest > testAclManagementMethods PASSED

kafka.zk.KafkaZkClientTest > testGetDataAndStat STARTED

kafka.zk.KafkaZkClientTest > testGetDataAndStat PASSED

kafka.zk.KafkaZkClientTest > testConditionalUpdatePath STARTED

kafka.zk.KafkaZkClientTest > testConditionalUpdatePath PASSED

kafka.zk.KafkaZkClientTest > testTopicAssignmentMethods STARTED

kafka.zk.KafkaZkClientTest > testTopicAssignmentMethods PASSED

kafka.zk.KafkaZkClientTest > testDeleteRecursive STARTED

kafka.zk.KafkaZkClientTest > testDeleteRecursive PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytesWithCompression 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytesWithCompression 
PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > 
testIteratorIsConsistentWithCompression STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > 
testIteratorIsConsistentWithCompression PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testIteratorIsConsistent 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testIteratorIsConsistent PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testEqualsWithCompression 
STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testEqualsWithCompression 
PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testWrittenEqualsRead STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testWrittenEqualsRead PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testEquals STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testEquals PASSED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytes STARTED

kafka.javaapi.message.ByteBufferMessageSetTest > testSizeInBytes PASSED

kafka.javaapi.consumer.ZookeeperConsumerConnectorTest > testBasic STARTED

kafka.javaapi.consumer.ZookeeperConsumerConnectorTest > testBasic PASSED

kafka.metrics.MetricsTest > testMetricsReporterAfterDeletingTopic STARTED

kafka.metrics.MetricsTest > testMetricsReporterAfterDeletingTopic PASSED

kafka.metrics.MetricsTest > 
testBrokerTopicMetricsUnregisteredAfterDeletingTopic STARTED

kafka.metrics.MetricsTest > 
testBrokerTopicMetricsUnregisteredAfterDeletingTopic PASSED

kafka.metrics.MetricsTest > testClusterIdMetric STARTED

kafka.metrics.MetricsTest > testClusterIdMetric PASSED

kafka.metrics.MetricsTest > testControllerMetrics STARTED

kafka.metrics.MetricsTest > testControllerMetrics PASSED

kafka.metrics.MetricsTest > testWindowsStyleTagNames STARTED

kafka.metrics.MetricsTest > testWindowsStyleTagNames PASSED

kafka.metrics.MetricsTest > testMetricsLeak STARTED

kafka.metrics.MetricsTest > testMetricsLeak PASSED

kafka.metrics.MetricsTest > testBrokerTopicMetricsBytesInOut STARTED

kafka.metrics.MetricsTest > testBrokerTopicMetricsBytesInOut PASSED

kafka.metrics.KafkaTimerTest > testKafkaTimer STARTED

kafka.metrics.KafkaTimerTest > testKafkaTimer PASSED

kafka.security.auth.ResourceTypeTest > testJavaConversions STARTED

kafka.security.auth.ResourceTypeTest > testJavaConversions PASSED

kafka.security.auth.ResourceTypeTest > testFromString STARTED

kafka.security.auth.ResourceTypeTest > testFromString PASSED

kafka.security.auth.OperationTest > testJavaConversions STARTED

kafka.security.auth.OperationTest > testJavaConversions PASSED

kafka.security.auth.ZkAuthorizationTest > testIsZkSecurityEnabled STARTED

kafka.security.auth.ZkAuthorizationTest > testIsZkSecurityEnabled PASSED

kafka.security.auth.ZkAuthorizationTest > testZkUtils STARTED

kafka.security.auth.ZkAuthorizationTest > testZkUtils PASSED

kafka.security.auth.ZkAuthorizationTest > tes

[GitHub] kafka pull request #4278: MINOR: fixed typo that causes incorrect values in ...

[kiest]

GitHub user kiest opened a pull request:

https://github.com/apache/kafka/pull/4278

MINOR: fixed typo that causes incorrect values in records-consumed-rate JMX 
metric

records-consumed-rate fetch metric shows different value than aggregated 
topic-level records-consumed-rate metrics. Looks like a typo when reporting 
bytes and records in Fetcher class.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/kiest/kafka trunk

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/4278.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #4278


commit bf19d4d8daae4122bcd5da75b481916c9e739ad1
Author: Igor Kostiakov 
Date:   2017-11-30T12:11:14Z

fixed typo that caused incorrect values in records-consumed-rate JMX metric




---

[jira] [Resolved] (KAFKA-6282) exactly_once semantics breaks demo application

[Romans Markuns (JIRA)]

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

Romans Markuns resolved KAFKA-6282.
---
Resolution: Not A Bug

> exactly_once semantics breaks demo application
> --
>
> Key: KAFKA-6282
> URL: https://issues.apache.org/jira/browse/KAFKA-6282
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Affects Versions: 0.11.0.0, 1.0.0
> Environment: Tested on i7+24GB Ubuntu16.04 and i7+8GB Windows 7, with 
> cluster 1.0.0 and 0.11.0.0 and streams 1.0.0 and 0.11.0.0
>Reporter: Romans Markuns
> Attachments: WordCountDemo.java, server.properties
>
>
> +What I try to achieve+
> Do successful run of Kafka streams app with setting "processing.guarantee" 
> set to "exactly_once"
> +How+
> Use Kafka quickstart example 
> (https://kafka.apache.org/10/documentation/streams/quickstart) and modify 
> only configuration parameters. 
> Things I've changed:
> 1) Add one line to WordCountDemo: 
> {code:java}
> props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, 
> StreamsConfig.EXACTLY_ONCE);
> {code}
> 2) Modify server.properties to be the same as we use in QA: set broker id to 
> 1, allow deleting topics via admin client and set initial rebalance delay to 
> 3 s.
> +What I expect+
> Modified demo app works exactly as the original as presented in link above.
> +What I get+
> 1) Original app works fine. Output topic after each line is submitted via 
> console producer.
> 2) Modified app does not process topic record after it is submitted via 
> console producer. Streams remain in state REBALANCING, no errors on warning 
> printed. MAIN thread forever blocks waiting TransactionCoordinator response 
> (CountdownLatch.await()) and this message getting printed: 
> [kafka-producer-network-thread | 
> streams-wordcount-client-StreamThread-1-0_0-producer] DEBUG 
> org.apache.kafka.clients.producer.internals.TransactionManager - [Producer 
> clientId=streams-wordcount-client-StreamThread-1-0_0-producer, 
> transactionalId=streams-wordcount-0_0] Enqueuing transactional request 
> (type=FindCoordinatorRequest, coordinatorKey=streams-wordcount-0_0, 
> coordinatorType=TRANSACTION)



--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

[GitHub] kafka pull request #4219: MINOR: Shutdown ControllerEventThread via event in...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/4219


---

Re: [VOTE] KIP-223 - Add per-topic min lead and per-partition lead metrics to KafkaConsumer

[charly molter]

+1 (non-binding)

On Thu, Nov 30, 2017 at 9:01 AM, Ted Yu  wrote:

> +1
>
> On Thu, Nov 30, 2017 at 12:56 AM, Manikumar 
> wrote:
>
> > +1 (non-binding)
> >
> > On Thu, Nov 30, 2017 at 8:35 AM, Hu Xi  wrote:
> >
> > > Hi all,
> > >
> > > As I didn't see any further discussion around this KIP, I'd like to
> start
> > > voting.
> > >
> > > KIP documentation:
> > >
> > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > > 223+-+Add+per-topic+min+lead+and+per-partition+lead+
> > > metrics+to+KafkaConsumer
> > >
> > >
> > >
> > > Cheers,
> > >
> > > huxihx
> > >
> >
>



-- 
Charly Molter

Re: [VOTE] KIP-223 - Add per-topic min lead and per-partition lead metrics to KafkaConsumer

[Ted Yu]

+1

On Thu, Nov 30, 2017 at 12:56 AM, Manikumar 
wrote:

> +1 (non-binding)
>
> On Thu, Nov 30, 2017 at 8:35 AM, Hu Xi  wrote:
>
> > Hi all,
> >
> > As I didn't see any further discussion around this KIP, I'd like to start
> > voting.
> >
> > KIP documentation:
> >
> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> > 223+-+Add+per-topic+min+lead+and+per-partition+lead+
> > metrics+to+KafkaConsumer
> >
> >
> >
> > Cheers,
> >
> > huxihx
> >
>

Re: [VOTE] KIP-223 - Add per-topic min lead and per-partition lead metrics to KafkaConsumer

[Manikumar]

+1 (non-binding)

On Thu, Nov 30, 2017 at 8:35 AM, Hu Xi  wrote:

> Hi all,
>
> As I didn't see any further discussion around this KIP, I'd like to start
> voting.
>
> KIP documentation:
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 223+-+Add+per-topic+min+lead+and+per-partition+lead+
> metrics+to+KafkaConsumer
>
>
>
> Cheers,
>
> huxihx
>